diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index d0e442b..cae6335 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -1,17 +1,18 @@ -# daemon/ produces the veloxd binary and the veloxd_rpc static library it is built from. +# daemon/ produces the veloxd binary and the static libraries it is built from. # Owned by lane DAEMON. Wired in by PKG via add_subdirectory(daemon) in the root file, # guarded on this file existing. # # Layering (CLAUDE.md §3): depends on velox::core and velox::proto. No Qt. The engine -# (velox::core) is not linked yet — it arrives when sched/ and the task glue land; this -# first drop is the RPC transport + dispatcher skeleton so the CLI and GUI have a real -# server to talk to. +# (velox::core) is not linked yet — it arrives when sched/ and the task glue land. This +# drop is the RPC transports (Unix socket + loopback WebSocket), the SQLite store, and a +# dispatcher skeleton so the CLI and GUI have a real server to talk to. if(NOT TARGET nlohmann_json::nlohmann_json) find_package(nlohmann_json 3.11 REQUIRED) endif() find_package(Threads REQUIRED) find_package(SQLite3 REQUIRED) +find_package(OpenSSL REQUIRED) # libcrypto: WebSocket accept hash, pairing token hash # --- generated: migrations_embedded.hpp from src/store/migrations/*.sql --------------- set(_mig_dir ${CMAKE_CURRENT_SOURCE_DIR}/src/store/migrations) @@ -26,10 +27,12 @@ add_custom_command( VERBATIM) add_custom_target(veloxd_migrations_hdr DEPENDS ${_mig_hdr}) -# --- veloxd_store — SQLite store + migrations ----------------------------------------- +# --- veloxd_store — SQLite store, migrations, crypto helpers -------------------------- add_library(veloxd_store STATIC + src/util/crypto.cpp src/store/sqlite.cpp src/store/migrations.cpp + src/store/pairings.cpp ${_mig_hdr} ) add_library(velox::daemon_store ALIAS veloxd_store) @@ -39,31 +42,33 @@ target_include_directories(veloxd_store ) target_compile_features(veloxd_store PUBLIC cxx_std_23) target_compile_options(veloxd_store PRIVATE -Wall -Wextra -Wpedantic -Werror) -target_link_libraries(veloxd_store PUBLIC SQLite::SQLite3) +target_link_libraries(veloxd_store PUBLIC SQLite::SQLite3 PRIVATE OpenSSL::Crypto) -# --- veloxd_rpc — the server library ---------------------------------------------------- +# --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------ add_library(veloxd_rpc STATIC src/rpc/runtime_dir.cpp src/rpc/event_loop.cpp src/rpc/uds_server.cpp + src/rpc/ws_frame.cpp + src/rpc/ws_handshake.cpp + src/rpc/ws_server.cpp + src/rpc/pairing.cpp src/rpc/dispatcher.cpp ) add_library(velox::daemon_rpc ALIAS veloxd_rpc) -target_include_directories(veloxd_rpc - PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src -) +target_include_directories(veloxd_rpc PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src) target_compile_features(veloxd_rpc PUBLIC cxx_std_23) target_compile_options(veloxd_rpc PRIVATE -Wall -Wextra -Wpedantic -Werror) target_link_libraries(veloxd_rpc - PUBLIC velox::proto nlohmann_json::nlohmann_json Threads::Threads + PUBLIC velox::proto veloxd_store nlohmann_json::nlohmann_json Threads::Threads ) # --- veloxd — the daemon binary ------------------------------------------------------- add_executable(veloxd src/main.cpp) target_compile_features(veloxd PRIVATE cxx_std_23) target_compile_options(veloxd PRIVATE -Wall -Wextra -Wpedantic -Werror) -target_link_libraries(veloxd PRIVATE veloxd_rpc veloxd_store) +target_link_libraries(veloxd PRIVATE veloxd_rpc) if(VELOX_BUILD_TESTS AND EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt) add_subdirectory(tests) diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index e410538..f8e5d11 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -1,8 +1,9 @@ // veloxd — the Velox download-manager daemon. // -// This drop wires up the Unix-socket RPC transport and a dispatcher skeleton so the CLI -// and GUI have a real server to speak to (AGENT-DAEMON.md build order, step 1). The -// WebSocket transport, the SQLite store and the scheduler land next. +// Wires up both RPC transports (Unix socket + loopback WebSocket), the SQLite store, +// and a dispatcher skeleton so the CLI and GUI have a real server to speak to +// (AGENT-DAEMON.md build order, steps 1 and 3). The scheduler and the engine link land +// next. #include #include @@ -16,8 +17,12 @@ #include "rpc/dispatcher.hpp" #include "rpc/event_loop.hpp" +#include "rpc/pairing.hpp" #include "rpc/runtime_dir.hpp" #include "rpc/uds_server.hpp" +#include "rpc/ws_server.hpp" +#include "store/migrations.hpp" +#include "store/sqlite.hpp" #include "version.hpp" namespace { @@ -80,15 +85,46 @@ int main() { ::sigaction(SIGTERM, &sa, nullptr); ::signal(SIGPIPE, SIG_IGN); // a client vanishing mid-write is EPIPE, never a signal + std::string data_dir; + if (const auto ec = velox::daemon::rpc::resolve_data_dir(data_dir)) { + std::cerr << "veloxd: cannot prepare data directory: " << ec.message() << "\n"; + return 1; + } + auto db = velox::daemon::store::Db::open(data_dir + "/velox.db"); + if (!db) { + std::cerr << "veloxd: cannot open " << data_dir << "/velox.db: " + << db.error().to_string() << "\n"; + return 1; + } + if (const auto m = velox::daemon::store::migrate_to_head(*db); !m) { + std::cerr << "veloxd: schema migration failed: " << m.error().to_string() << "\n"; + return 1; + } + velox::daemon::rpc::VeloxDispatcher dispatcher; + velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path()); if (const auto ec = uds.start()) { std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message() << "\n"; return 1; } - std::cout << "veloxd: listening on " << uds.socket_path() << "\n"; + + // The WebSocket transport is the extension's fallback (docs/05 §4); the Unix socket is + // the primary. If every port in 52000-52016 is taken, log it and carry on rather than + // refusing to start — capture must fail open, and the GUI/CLI still have the socket. + // TODO(build step 7): replace EnvAutoApprover with a GUI-dialog / desktop-notification + // approver. Until then pairing needs VELOX_PAIR_AUTO=1. + velox::daemon::rpc::EnvAutoApprover approver; + velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, rt); + if (const auto ec = ws.start()) { + std::cerr << "veloxd: WebSocket transport unavailable (" << ec.message() + << "); the extension fallback will not work this run\n"; + } else { + std::cout << "veloxd: WebSocket transport on 127.0.0.1:" << ws.port() << "\n"; + } + loop.run(); std::cout << "veloxd: shutting down\n"; diff --git a/daemon/src/rpc/pairing.cpp b/daemon/src/rpc/pairing.cpp new file mode 100644 index 0000000..6390d0c --- /dev/null +++ b/daemon/src/rpc/pairing.cpp @@ -0,0 +1,54 @@ +#include "rpc/pairing.hpp" + +#include +#include +#include + +namespace velox::daemon::rpc { + +bool EnvAutoApprover::approve(const PairingRequest& req) { + (void)req; + const char* v = std::getenv("VELOX_PAIR_AUTO"); + return v != nullptr && std::string_view(v) == "1"; +} + +PairingRateLimiter::Decision PairingRateLimiter::check(std::string_view origin, + Clock::time_point now) { + auto it = by_origin_.find(origin); + if (it == by_origin_.end()) return {true, 0}; + + Entry& e = it->second; + if (now < e.locked_until) { + const auto left = + std::chrono::duration_cast(e.locked_until - now).count(); + return {false, static_cast(left) + 1}; + } + + while (!e.failures.empty() && now - e.failures.front() > kWindow) e.failures.pop_front(); + if (static_cast(e.failures.size()) >= kMaxPerWindow) { + e.locked_until = now + kLockout; + return {false, static_cast(kLockout.count())}; + } + return {true, 0}; +} + +void PairingRateLimiter::record_failure(std::string_view origin, Clock::time_point now) { + Entry& e = by_origin_.try_emplace(std::string(origin)).first->second; + while (!e.failures.empty() && now - e.failures.front() > kWindow) e.failures.pop_front(); + e.failures.push_back(now); + if (static_cast(e.failures.size()) >= kMaxPerWindow) e.locked_until = now + kLockout; +} + +void PairingRateLimiter::record_success(std::string_view origin) { + by_origin_.erase(std::string(origin)); +} + +std::string make_pairing_code() { + std::random_device rd; + std::uniform_int_distribution d(0, 9999); + char buf[5]; + std::snprintf(buf, sizeof(buf), "%04d", d(rd)); + return std::string(buf); +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/pairing.hpp b/daemon/src/rpc/pairing.hpp new file mode 100644 index 0000000..bdcaab6 --- /dev/null +++ b/daemon/src/rpc/pairing.hpp @@ -0,0 +1,71 @@ +#pragma once + +// The human side of session.pair: showing the user a code and getting an Allow / Deny. +// +// docs/05 §4 wants a GUI dialog when the GUI is connected, else a desktop notification +// with actions. Neither exists yet (that is integration, build step 7), so this is an +// interface with a development stub. The token mechanism around it — generation, hashing, +// storage, revocation, rate limiting — is real. + +#include +#include +#include +#include +#include +#include + +namespace velox::daemon::rpc { + +struct PairingRequest { + std::string origin; // moz-extension://, from the verified Origin header + std::string client_name; // SessionPairParams.clientName, shown in the prompt + std::string code; // four digits, shown to the user and echoable in Options +}; + +class PairingApprover { +public: + virtual ~PairingApprover() = default; + // Returns true iff the user approved. Must not block the RPC loop indefinitely; the + // real notification-backed approver will run async and is not this shape. + virtual bool approve(const PairingRequest& req) = 0; +}; + +// Development / test stub: approves iff $VELOX_PAIR_AUTO == "1", otherwise denies. Never +// shipped as the default in a release build. +class EnvAutoApprover final : public PairingApprover { +public: + bool approve(const PairingRequest& req) override; +}; + +// Per-origin failed-attempt limiter: 5 failures in a rolling 60 s, then a 60 s lockout +// (docs/05 §4, fixture session.pair.rate-limited). In-memory and keyed by origin, so a +// reconnect does not reset it. A success clears the origin's history. +class PairingRateLimiter { +public: + using Clock = std::chrono::steady_clock; + + struct Decision { + bool allowed; + int retry_after_sec; // set when !allowed + }; + + Decision check(std::string_view origin, Clock::time_point now = Clock::now()); + void record_failure(std::string_view origin, Clock::time_point now = Clock::now()); + void record_success(std::string_view origin); + +private: + static constexpr int kMaxPerWindow = 5; + static constexpr auto kWindow = std::chrono::seconds(60); + static constexpr auto kLockout = std::chrono::seconds(60); + + struct Entry { + std::deque failures; + Clock::time_point locked_until{}; + }; + std::map> by_origin_; +}; + +// A four-digit code for the prompt. Uniform over 0000-9999. +std::string make_pairing_code(); + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/runtime_dir.cpp b/daemon/src/rpc/runtime_dir.cpp index e387916..d293411 100644 --- a/daemon/src/rpc/runtime_dir.cpp +++ b/daemon/src/rpc/runtime_dir.cpp @@ -53,4 +53,24 @@ std::error_code resolve_runtime_dir(RuntimeDir& out) { return {}; } +std::error_code resolve_data_dir(std::string& out) { + std::string base; + if (const char* xdg = ::getenv("XDG_DATA_HOME"); xdg != nullptr && xdg[0] != '\0') { + base = xdg; + } else if (const char* home = ::getenv("HOME"); home != nullptr && home[0] != '\0') { + base = std::string(home) + "/.local/share"; + } else { + return errc(ENOENT); + } + if (!base.empty() && base.back() == '/') base.pop_back(); + + // Create the XDG base components leniently, then the velox dir with a strict check. + ::mkdir(base.c_str(), 0700); + const std::string dir = base + "/velox"; + if (auto ec = ensure_private_dir(dir)) return ec; + + out = dir; + return {}; +} + } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/runtime_dir.hpp b/daemon/src/rpc/runtime_dir.hpp index b71a529..3b4f5ab 100644 --- a/daemon/src/rpc/runtime_dir.hpp +++ b/daemon/src/rpc/runtime_dir.hpp @@ -27,4 +27,9 @@ struct RuntimeDir { // owner, wrong perms, mkdir failed). std::error_code resolve_runtime_dir(RuntimeDir& out); +// The persistent data directory: $XDG_DATA_HOME/velox or ~/.local/share/velox +// (docs/01 §5). Created 0700 if absent. Holds velox.db. On success `out` is the absolute +// path with no trailing slash. +std::error_code resolve_data_dir(std::string& out); + } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_frame.cpp b/daemon/src/rpc/ws_frame.cpp new file mode 100644 index 0000000..69a1b09 --- /dev/null +++ b/daemon/src/rpc/ws_frame.cpp @@ -0,0 +1,152 @@ +#include "rpc/ws_frame.hpp" + +#include + +namespace velox::daemon::rpc { + +namespace { + +bool is_control(WsOpcode op) { + return op == WsOpcode::Close || op == WsOpcode::Ping || op == WsOpcode::Pong; +} +bool is_known_data(WsOpcode op) { + return op == WsOpcode::Text || op == WsOpcode::Binary; +} + +} // namespace + +WsFrameReader::Status WsFrameReader::feed(std::string_view bytes, + std::vector& messages) { + buf_.append(bytes); + + for (;;) { + if (buf_.size() < 2) return Status::Ok; + + const auto b0 = static_cast(buf_[0]); + const auto b1 = static_cast(buf_[1]); + + const bool fin = (b0 & 0x80) != 0; + const std::uint8_t rsv = b0 & 0x70; + const auto opcode = static_cast(b0 & 0x0F); + const bool masked = (b1 & 0x80) != 0; + std::uint64_t len = b1 & 0x7F; + + if (rsv != 0) { + error_ = "RSV bits set with no negotiated extension"; + return Status::ProtocolError; + } + if (!masked) { + error_ = "client frame is not masked"; // RFC 6455 §5.1 + return Status::ProtocolError; + } + + std::size_t header = 2; + if (len == 126) { + if (buf_.size() < 4) return Status::Ok; + len = (static_cast(static_cast(buf_[2])) << 8) | + static_cast(buf_[3]); + header = 4; + } else if (len == 127) { + if (buf_.size() < 10) return Status::Ok; + len = 0; + for (int i = 0; i < 8; ++i) + len = (len << 8) | static_cast(buf_[2 + i]); + header = 10; + } + + if (is_control(opcode)) { + if (!fin) { + error_ = "fragmented control frame"; + return Status::ProtocolError; + } + if (len > 125) { + error_ = "control frame payload over 125 bytes"; + return Status::ProtocolError; + } + } + if (len > kMaxMessageBytes || frag_.size() + len > kMaxMessageBytes) { + error_ = "message exceeds the size cap"; + return Status::MessageTooBig; + } + + const std::size_t need = header + 4 + static_cast(len); + if (buf_.size() < need) return Status::Ok; + + const char* mask = buf_.data() + header; + const char* body = mask + 4; + + std::string payload; + payload.resize(static_cast(len)); + for (std::uint64_t i = 0; i < len; ++i) + payload[i] = static_cast(body[i] ^ mask[i & 3]); + + buf_.erase(0, need); + + // --- dispatch by opcode --------------------------------------------------- + if (is_control(opcode)) { + messages.push_back(WsMessage{opcode, std::move(payload)}); + continue; + } + + if (opcode == WsOpcode::Continuation) { + if (!in_fragment_) { + error_ = "continuation frame with nothing to continue"; + return Status::ProtocolError; + } + frag_.insert(frag_.end(), payload.begin(), payload.end()); + if (fin) { + messages.push_back( + WsMessage{frag_opcode_, std::string(frag_.begin(), frag_.end())}); + frag_.clear(); + in_fragment_ = false; + } + continue; + } + + if (!is_known_data(opcode)) { + error_ = "unknown opcode"; + return Status::ProtocolError; + } + if (in_fragment_) { + error_ = "new data frame started mid-fragment"; + return Status::ProtocolError; + } + if (fin) { + messages.push_back(WsMessage{opcode, std::move(payload)}); + } else { + frag_opcode_ = opcode; + in_fragment_ = true; + frag_.assign(payload.begin(), payload.end()); + } + } +} + +std::string ws_encode(WsOpcode opcode, std::string_view payload) { + std::string out; + out.push_back(static_cast(0x80 | static_cast(opcode))); // FIN + opcode + + const std::size_t n = payload.size(); + if (n < 126) { + out.push_back(static_cast(n)); + } else if (n <= 0xFFFF) { + out.push_back(static_cast(126)); + out.push_back(static_cast((n >> 8) & 0xFF)); + out.push_back(static_cast(n & 0xFF)); + } else { + out.push_back(static_cast(127)); + for (int i = 7; i >= 0; --i) + out.push_back(static_cast((static_cast(n) >> (i * 8)) & 0xFF)); + } + out.append(payload); // server frames are never masked + return out; +} + +std::string ws_close_payload(std::uint16_t code, std::string_view reason) { + std::string p; + p.push_back(static_cast((code >> 8) & 0xFF)); + p.push_back(static_cast(code & 0xFF)); + p.append(reason); + return p; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_frame.hpp b/daemon/src/rpc/ws_frame.hpp new file mode 100644 index 0000000..abe281a --- /dev/null +++ b/daemon/src/rpc/ws_frame.hpp @@ -0,0 +1,61 @@ +#pragma once + +// RFC 6455 frame codec — the security-sensitive parser on the loopback WebSocket +// transport. Incremental: feed() takes whatever bytes arrived and yields whole messages. +// A client frame MUST be masked (RFC 6455 §5.1); an unmasked client frame is a protocol +// error and the caller must close 1002. +// +// Kept deliberately small: text and binary data frames (reassembled across continuation +// frames), plus ping / pong / close control frames. No extensions, no RSV bits. + +#include +#include +#include +#include + +namespace velox::daemon::rpc { + +enum class WsOpcode : std::uint8_t { + Continuation = 0x0, + Text = 0x1, + Binary = 0x2, + Close = 0x8, + Ping = 0x9, + Pong = 0xA, +}; + +struct WsMessage { + WsOpcode opcode; // Text, Binary, Close, Ping or Pong (never Continuation) + std::string payload; // reassembled; unmasked +}; + +class WsFrameReader { +public: + enum class Status { Ok, ProtocolError, MessageTooBig }; + + // Append `bytes` and pull out every message they complete. On a non-Ok status the + // caller sends a Close and drops the connection; `messages` still holds anything + // decoded before the fault. + Status feed(std::string_view bytes, std::vector& messages); + + std::string_view error() const noexcept { return error_; } + +private: + // Cap on a single reassembled message. A JSON-RPC call over this transport is small; + // past this the peer is misbehaving. + static constexpr std::size_t kMaxMessageBytes = 8 * 1024 * 1024; + + std::string buf_; // undecoded bytes + std::vector frag_; // partial data message across continuations + WsOpcode frag_opcode_ = WsOpcode::Text; + bool in_fragment_ = false; + std::string error_; +}; + +// Build a server->client frame (never masked). `payload` may be empty for Close/Ping/Pong. +std::string ws_encode(WsOpcode opcode, std::string_view payload); + +// A Close frame body: 2-byte big-endian status code, optional UTF-8 reason. +std::string ws_close_payload(std::uint16_t code, std::string_view reason = {}); + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_handshake.cpp b/daemon/src/rpc/ws_handshake.cpp new file mode 100644 index 0000000..a4dc6df --- /dev/null +++ b/daemon/src/rpc/ws_handshake.cpp @@ -0,0 +1,121 @@ +#include "rpc/ws_handshake.hpp" + +#include +#include +#include +#include +#include + +#include "util/crypto.hpp" + +namespace velox::daemon::rpc { + +namespace { + +constexpr std::string_view kGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + +std::string lower(std::string_view s) { + std::string out(s); + std::transform(out.begin(), out.end(), out.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return out; +} + +std::string_view trim(std::string_view s) { + while (!s.empty() && (s.front() == ' ' || s.front() == '\t')) s.remove_prefix(1); + while (!s.empty() && (s.back() == ' ' || s.back() == '\t' || s.back() == '\r')) + s.remove_suffix(1); + return s; +} + +std::string simple_response(std::string_view status_line, std::string_view body) { + std::string r; + r.append("HTTP/1.1 ").append(status_line).append("\r\n"); + r.append("Content-Length: ").append(std::to_string(body.size())).append("\r\n"); + r.append("Connection: close\r\n\r\n"); + r.append(body); + return r; +} + +bool looks_like_extension_origin(std::string_view origin) { + // moz-extension://. We do not pin a specific extension id here — the + // pairing token is the identity; this only rejects a page origin (http/https/file). + return origin.rfind("moz-extension://", 0) == 0 && origin.size() > 16; +} + +} // namespace + +std::string ws_accept_key(std::string_view sec_websocket_key) { + std::string concat(sec_websocket_key); + concat.append(kGuid); + const auto digest = velox::daemon::crypto::sha1(concat); + return velox::daemon::crypto::base64_encode(digest.data(), digest.size()); +} + +HandshakeResult ws_try_handshake(std::string_view buffer) { + HandshakeResult res; + + const auto end = buffer.find("\r\n\r\n"); + if (end == std::string_view::npos) return res; // headers still arriving + res.complete = true; + res.consumed = end + 4; + + const std::string_view head = buffer.substr(0, end); + const auto first_nl = head.find("\r\n"); + const std::string_view request_line = head.substr(0, first_nl); + + std::map headers; + std::size_t pos = (first_nl == std::string_view::npos) ? head.size() : first_nl + 2; + while (pos < head.size()) { + const auto nl = head.find("\r\n", pos); + const std::string_view line = + head.substr(pos, nl == std::string_view::npos ? head.size() - pos : nl - pos); + const auto colon = line.find(':'); + if (colon != std::string_view::npos) { + headers[lower(trim(line.substr(0, colon)))] = + std::string(trim(line.substr(colon + 1))); + } + if (nl == std::string_view::npos) break; + pos = nl + 2; + } + + auto get = [&](const char* k) -> std::string_view { + const auto it = headers.find(k); + return it == headers.end() ? std::string_view{} : std::string_view{it->second}; + }; + + const bool is_get = request_line.rfind("GET ", 0) == 0; + const bool upgrade_ws = lower(get("upgrade")).find("websocket") != std::string::npos; + const bool conn_upgrade = lower(get("connection")).find("upgrade") != std::string::npos; + const std::string_view key = get("sec-websocket-key"); + const std::string_view version = get("sec-websocket-version"); + const std::string_view origin = get("origin"); + + if (!is_get || !upgrade_ws || !conn_upgrade || key.empty()) { + res.response = simple_response("400 Bad Request", "not a WebSocket upgrade"); + return res; + } + if (version != "13") { + std::string r = "HTTP/1.1 426 Upgrade Required\r\nSec-WebSocket-Version: 13\r\n"; + r.append("Connection: close\r\n\r\n"); + res.response = std::move(r); + return res; + } + if (origin.empty() || !looks_like_extension_origin(origin)) { + // docs/05 §4: verify Origin is a moz-extension origin. A page cannot pair. + res.response = simple_response("403 Forbidden", "origin not permitted"); + return res; + } + + std::string r = "HTTP/1.1 101 Switching Protocols\r\n"; + r.append("Upgrade: websocket\r\n"); + r.append("Connection: Upgrade\r\n"); + r.append("Sec-WebSocket-Accept: ").append(ws_accept_key(key)).append("\r\n\r\n"); + + res.ok = true; + res.response = std::move(r); + res.origin = std::string(origin); + return res; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_handshake.hpp b/daemon/src/rpc/ws_handshake.hpp new file mode 100644 index 0000000..c54c6a7 --- /dev/null +++ b/daemon/src/rpc/ws_handshake.hpp @@ -0,0 +1,30 @@ +#pragma once + +// The RFC 6455 opening handshake, plus the two checks the extension spec makes +// non-negotiable (docs/05 §4): the request must carry an Origin, and it must look like a +// Firefox extension origin (moz-extension://). The token check happens later, in +// session.hello — the handshake only gets the socket to WebSocket framing. + +#include +#include + +namespace velox::daemon::rpc { + +struct HandshakeResult { + bool complete = false; // a full request was parsed + bool ok = false; // ... and it is a valid, allowed upgrade + std::string response; // bytes to write back: 101 on ok, 400/403 otherwise + std::string origin; // the verified Origin, when ok + std::size_t consumed = 0; // bytes of input that formed the request +}; + +// Parse an accumulating HTTP request buffer. Returns complete=false (and consumed=0) while +// the header block is still arriving. Once "\r\n\r\n" is seen, validates and fills in the +// 101 (or an error) response. A body, if any, is not expected on an upgrade and is +// ignored. +HandshakeResult ws_try_handshake(std::string_view buffer); + +// Exposed for the unit test: RFC 6455 §1.3 accept value for a client key. +std::string ws_accept_key(std::string_view sec_websocket_key); + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp new file mode 100644 index 0000000..d6c701c --- /dev/null +++ b/daemon/src/rpc/ws_server.cpp @@ -0,0 +1,440 @@ +#include "rpc/ws_server.hpp" + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +#include "rpc/event_loop.hpp" +#include "rpc/ws_handshake.hpp" +#include "store/pairings.hpp" +#include "store/sqlite.hpp" +#include "version.hpp" + +namespace velox::daemon::rpc { + +namespace proto = velox::proto; +using nlohmann::json; + +namespace { + +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 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 s[37]; + std::snprintf(s, sizeof(s), "%08x-%04x-%04x-%04x-%04x%08x", a, (b >> 16), (b & 0xFFFF), + (c >> 16), (c & 0xFFFF), e); + return std::string(s); +} + +int major_of(const std::string& semver) { + try { + return std::stoi(semver.substr(0, semver.find('.'))); + } catch (...) { + return -1; + } +} + +json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json data = nullptr) { + return proto::make_error(id, code, msg, std::move(data)); +} + +} // namespace + +WsServer::WsServer(EventLoop& loop, proto::Dispatcher& dispatcher, store::Db& db, + PairingApprover& approver, RuntimeDir runtime) + : loop_(loop), + dispatcher_(dispatcher), + db_(db), + approver_(approver), + runtime_(std::move(runtime)) {} + +WsServer::~WsServer() { + for (auto& [fd, c] : conns_) { + loop_.del_fd(fd); + ::close(fd); + } + if (listen_fd_ >= 0) { + loop_.del_fd(listen_fd_); + ::close(listen_fd_); + } + if (wrote_port_file_) ::unlink(runtime_.ws_port_path().c_str()); +} + +std::error_code WsServer::start() { + int fd = -1; + for (int p = kPortLo; p <= kPortHi; ++p) { + fd = ::socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (fd < 0) return errc(errno); + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK); // 127.0.0.1 only — never INADDR_ANY + addr.sin_port = ::htons(static_cast(p)); + + if (::bind(fd, reinterpret_cast(&addr), sizeof(addr)) == 0) { + port_ = p; + break; + } + ::close(fd); + fd = -1; + if (errno != EADDRINUSE) return errc(errno); + } + if (fd < 0) return errc(EADDRINUSE); // 52000-52016 all taken + + if (::listen(fd, SOMAXCONN) != 0) { + const int e = errno; + ::close(fd); + return errc(e); + } + + // Publish the port for the extension, which cannot read $XDG_RUNTIME_DIR itself but + // can be told where to look by a connected GUI. 0600, same as the socket. + const std::string pf = runtime_.ws_port_path(); + const int pfd = ::open(pf.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0600); + if (pfd < 0) { + const int e = errno; + ::close(fd); + return errc(e); + } + const std::string line = std::to_string(port_) + "\n"; + [[maybe_unused]] ssize_t w = ::write(pfd, line.data(), line.size()); + ::close(pfd); + wrote_port_file_ = true; + + listen_fd_ = fd; + loop_.add_fd(listen_fd_, kRead, [this](int, unsigned) { on_listener_readable(); }); + return {}; +} + +void WsServer::on_listener_readable() { + for (;;) { + const int cfd = ::accept4(listen_fd_, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); + if (cfd < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + if (errno == EINTR || errno == ECONNABORTED) continue; + break; + } + auto conn = std::make_unique(); + conn->fd = cfd; + conns_.emplace(cfd, std::move(conn)); + loop_.add_fd(cfd, kRead, [this](int fd, unsigned ev) { on_conn_event(fd, ev); }); + } +} + +void WsServer::on_conn_event(int fd, unsigned events) { + const auto it = conns_.find(fd); + if (it == conns_.end()) return; + Conn& c = *it->second; + + if (events & kWrite) { + flush(c); + if (conns_.find(fd) == conns_.end()) return; + } + if (!(events & kRead)) return; + + char buf[64 * 1024]; + for (;;) { + const ssize_t n = ::read(fd, buf, sizeof(buf)); + if (n > 0) { + const std::string_view chunk(buf, static_cast(n)); + if (c.phase == Phase::Handshake) { + c.in_raw.append(chunk); + if (c.in_raw.size() > kMaxHandshakeBytes) { + close_conn(fd); + return; + } + progress_handshake(c); + } else { + on_ws_bytes(c, chunk); + } + if (conns_.find(fd) == conns_.end()) return; + continue; + } + if (n == 0) { + close_conn(fd); + return; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + if (errno == EINTR) continue; + close_conn(fd); + return; + } +} + +void WsServer::progress_handshake(Conn& c) { + const HandshakeResult hs = ws_try_handshake(c.in_raw); + if (!hs.complete) return; + + c.outbuf.append(hs.response); + if (!hs.ok) { + c.close_after_flush = true; + flush(c); + return; + } + + c.origin = hs.origin; + c.phase = Phase::Open; + std::string leftover = c.in_raw.substr(hs.consumed); + c.in_raw.clear(); + flush(c); + if (conns_.count(c.fd) && !leftover.empty()) on_ws_bytes(c, leftover); +} + +void WsServer::on_ws_bytes(Conn& c, std::string_view bytes) { + std::vector msgs; + const auto st = c.frames.feed(bytes, msgs); + + for (auto& m : msgs) { + switch (m.opcode) { + case WsOpcode::Text: + handle_rpc(c, m.payload); + break; + case WsOpcode::Binary: + begin_close(c, 1003, "binary frames are not accepted"); // 1003: unacceptable data + break; + case WsOpcode::Ping: + send_frame(c, WsOpcode::Pong, m.payload); + break; + case WsOpcode::Pong: + break; + case WsOpcode::Close: + send_frame(c, WsOpcode::Close, m.payload); + c.close_after_flush = true; + flush(c); + return; + default: + break; + } + if (conns_.find(c.fd) == conns_.end()) return; + } + + if (st == WsFrameReader::Status::ProtocolError) { + begin_close(c, 1002, c.frames.error()); // 1002: protocol error + } else if (st == WsFrameReader::Status::MessageTooBig) { + begin_close(c, 1009, "message too big"); // 1009: message too big + } +} + +void WsServer::handle_rpc(Conn& c, const std::string& text) { + json req = json::parse(text, nullptr, false); + if (req.is_discarded()) { + send_text(c, rpc_error(nullptr, proto::ErrorCode::ParseError, "invalid JSON")); + return; + } + const json id = req.is_object() && req.contains("id") ? req.at("id") : json(nullptr); + const std::string method = + req.is_object() && req.contains("method") && req.at("method").is_string() + ? req.at("method").get() + : std::string{}; + + if (!method.empty()) { + json reply; + if (handle_session_ws(c, method, req, reply)) { + if (!reply.is_null()) send_text(c, reply); + return; + } + } + + // Any non-session method requires an authenticated connection. + if (!c.authed) { + send_text(c, rpc_error(id, proto::ErrorCode::NotPaired, "not paired: call session.pair first")); + return; + } + + json reply = proto::dispatch(dispatcher_, proto::Transport::Ws, req); + if (!reply.is_null()) send_text(c, reply); +} + +bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json& request, + json& reply) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params = request.contains("params") ? request.at("params") : json::object(); + + if (method == "session.pair") { + const auto dec = rate_limiter_.check(c.origin); + if (!dec.allowed) { + reply = rpc_error(id, proto::ErrorCode::RateLimited, + "too many pairing attempts; try again later", + json{{"retryAfterSec", dec.retry_after_sec}}); + return true; + } + auto p = proto::parse(params, "params"); + if (!p) { + reply = rpc_error(id, proto::ErrorCode::InvalidParams, p.error().message, + json{{"path", p.error().path}}); + return true; + } + + PairingRequest pr{c.origin, p->clientName, make_pairing_code()}; + if (!approver_.approve(pr)) { + rate_limiter_.record_failure(c.origin); + reply = rpc_error(id, proto::ErrorCode::NotPaired, "pairing was not approved"); + return true; + } + + store::Pairings pairings(db_); + auto created = pairings.create(c.origin, p->clientName, now_iso()); + if (!created) { + reply = rpc_error(id, proto::ErrorCode::InternalError, + "could not store the pairing: " + created.error().message); + return true; + } + rate_limiter_.record_success(c.origin); + + proto::SessionPairResult r; + r.token = created->token; + reply = proto::make_result(id, r); + return true; + } + + if (method == "session.hello") { + auto p = proto::parse(params, "params"); + if (!p) { + reply = rpc_error(id, proto::ErrorCode::InvalidParams, p.error().message, + json{{"path", p.error().path}}); + return true; + } + const int want = major_of(std::string(proto::kProtocolVersion)); + const int got = major_of(p->protocolVersion); + if (got != want) { + reply = rpc_error(id, proto::ErrorCode::VersionMismatch, + "protocol major version mismatch", + json{{"expected", std::string(proto::kProtocolVersion)}, + {"actual", p->protocolVersion}}); + c.close_after_flush = true; + return true; + } + + store::Pairings pairings(db_); + auto found = p->token ? pairings.find_active_by_token(*p->token) + : store::DbResult>(std::nullopt); + if (!found) { + reply = rpc_error(id, proto::ErrorCode::InternalError, found.error().message); + return true; + } + if (!found->has_value()) { + // Absent, malformed or wrong — all reported the same, and all count toward the + // pairing rate limit (fixture session.hello.not-paired). + rate_limiter_.record_failure(c.origin); + reply = rpc_error(id, proto::ErrorCode::NotPaired, "not paired: call session.pair first"); + return true; + } + + c.authed = true; + c.pairing_id = (*found)->pairing_id; + if (c.session_id.empty()) c.session_id = uuid4(); + (void)pairings.touch(c.pairing_id, now_iso()); + + proto::SessionHelloResult r; + r.daemonVersion = std::string(velox::daemon::kDaemonVersion); + r.protocolVersion = std::string(proto::kProtocolVersion); + r.sessionId = c.session_id; + r.transport = proto::SessionHelloResultTransport::Ws; + reply = proto::make_result(id, r); + return true; + } + + if (method == "session.subscribe") { + if (!c.authed) { + reply = rpc_error(id, proto::ErrorCode::NotPaired, "not paired: call session.pair first"); + return true; + } + auto p = proto::parse(params, "params"); + if (!p) { + reply = rpc_error(id, proto::ErrorCode::InvalidParams, p.error().message, + json{{"path", p.error().path}}); + return true; + } + proto::SessionSubscribeResult r; + r.ok = true; + for (const auto& ev : p->events) r.events.emplace_back(proto::to_string(ev)); + reply = proto::make_result(id, r); + return true; + } + + return false; +} + +void WsServer::send_text(Conn& c, const json& value) { + send_frame(c, WsOpcode::Text, value.dump()); +} + +void WsServer::send_frame(Conn& c, WsOpcode op, std::string_view payload) { + c.outbuf += ws_encode(op, payload); + if (c.outbuf.size() - c.out_off > kMaxOutBytes) { + close_conn(c.fd); + return; + } + flush(c); +} + +void WsServer::begin_close(Conn& c, std::uint16_t code, std::string_view reason) { + if (c.phase == Phase::Closing) return; + c.phase = Phase::Closing; + send_frame(c, WsOpcode::Close, ws_close_payload(code, reason)); + if (conns_.count(c.fd)) { + c.close_after_flush = true; + flush(c); + } +} + +void WsServer::flush(Conn& c) { + while (c.out_off < c.outbuf.size()) { + const ssize_t n = ::write(c.fd, c.outbuf.data() + c.out_off, c.outbuf.size() - c.out_off); + if (n > 0) { + c.out_off += static_cast(n); + continue; + } + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + loop_.mod_fd(c.fd, kRead | kWrite); + return; + } + if (n < 0 && errno == EINTR) continue; + close_conn(c.fd); + return; + } + c.outbuf.clear(); + c.out_off = 0; + if (c.close_after_flush) { + close_conn(c.fd); + return; + } + loop_.mod_fd(c.fd, kRead); +} + +void WsServer::close_conn(int fd) { + if (const auto it = conns_.find(fd); it != conns_.end()) { + loop_.del_fd(fd); + ::close(fd); + conns_.erase(it); + } +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_server.hpp b/daemon/src/rpc/ws_server.hpp new file mode 100644 index 0000000..72e467a --- /dev/null +++ b/daemon/src/rpc/ws_server.hpp @@ -0,0 +1,92 @@ +#pragma once + +// The loopback WebSocket transport (docs/05 §4). Binds 127.0.0.1 only, on the first free +// port in 52000-52016, and writes the chosen port to /ws.port. Any local process +// can connect, so a token is mandatory: session.pair mints one (behind a user prompt, +// rate-limited), session.hello must present it, and every other method is refused -32002 +// until it does. Privileged methods are refused -32003 by the generated dispatch(). + +#include +#include +#include +#include +#include + +#include + +#include "rpc/pairing.hpp" +#include "rpc/runtime_dir.hpp" +#include "rpc/ws_frame.hpp" +#include "velox_proto.hpp" + +namespace velox::daemon::store { +class Db; +} + +namespace velox::daemon::rpc { + +class EventLoop; + +class WsServer { +public: + WsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, velox::daemon::store::Db& db, + PairingApprover& approver, RuntimeDir runtime); + ~WsServer(); + + WsServer(const WsServer&) = delete; + WsServer& operator=(const WsServer&) = delete; + + // Pick a port, bind loopback, write ws.port, listen, register with the loop. + std::error_code start(); + + int port() const noexcept { return port_; } + std::size_t connection_count() const noexcept { return conns_.size(); } + + static constexpr int kPortLo = 52000; + static constexpr int kPortHi = 52016; + +private: + enum class Phase { Handshake, Open, Closing }; + + struct Conn { + int fd; + Phase phase = Phase::Handshake; + std::string in_raw; // bytes before the upgrade completes + WsFrameReader frames; + std::string outbuf; + std::size_t out_off = 0; + bool close_after_flush = false; + std::string origin; + bool authed = false; + std::string pairing_id; + std::string session_id; + }; + + void on_listener_readable(); + void on_conn_event(int fd, unsigned events); + void progress_handshake(Conn& c); + void on_ws_bytes(Conn& c, std::string_view bytes); + void handle_rpc(Conn& c, const std::string& text); + bool handle_session_ws(Conn& c, const std::string& method, const nlohmann::json& request, + nlohmann::json& reply); + + void send_text(Conn& c, const nlohmann::json& value); + void send_frame(Conn& c, WsOpcode op, std::string_view payload); + void begin_close(Conn& c, std::uint16_t code, std::string_view reason); + void flush(Conn& c); + void close_conn(int fd); + + EventLoop& loop_; + velox::proto::Dispatcher& dispatcher_; + velox::daemon::store::Db& db_; + PairingApprover& approver_; + RuntimeDir runtime_; + PairingRateLimiter rate_limiter_; + + int listen_fd_ = -1; + int port_ = 0; + bool wrote_port_file_ = false; + std::unordered_map> conns_; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/store/pairings.cpp b/daemon/src/store/pairings.cpp new file mode 100644 index 0000000..743a17e --- /dev/null +++ b/daemon/src/store/pairings.cpp @@ -0,0 +1,104 @@ +#include "store/pairings.hpp" + +#include + +#include + +#include "util/crypto.hpp" + +namespace velox::daemon::store { + +namespace { + +std::string uuid4() { + std::random_device rd; + std::uniform_int_distribution 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); +} + +Pairing read_row(Stmt& s) { + Pairing p; + p.pairing_id = s.column_text(0); + p.origin = s.column_text(1); + p.label = s.column_text(2); + p.created_at = s.column_text(3); + if (!s.column_is_null(4)) p.last_seen_at = s.column_text(4); + if (!s.column_is_null(5)) p.revoked_at = s.column_text(5); + return p; +} + +constexpr std::string_view kCols = + "pairing_id, origin, label, created_at, last_seen_at, revoked_at"; + +} // namespace + +DbResult Pairings::create(std::string_view origin, std::string_view label, + std::string_view now_iso) { + Created out{uuid4(), velox::daemon::crypto::random_token(32)}; + const std::string hash = velox::daemon::crypto::sha256_hex(out.token); + + auto st = db_.prepare( + "INSERT INTO pairings(pairing_id, token_sha256, origin, label, created_at) " + "VALUES(?1, ?2, ?3, ?4, ?5)"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, out.pairing_id); !r) return std::unexpected(r.error()); + if (auto r = st->bind(2, std::string_view(hash)); !r) return std::unexpected(r.error()); + if (auto r = st->bind(3, origin); !r) return std::unexpected(r.error()); + if (auto r = st->bind(4, label); !r) return std::unexpected(r.error()); + if (auto r = st->bind(5, now_iso); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return out; +} + +DbResult> Pairings::find_active_by_token(std::string_view token) { + const std::string hash = velox::daemon::crypto::sha256_hex(token); + auto st = db_.prepare(std::string("SELECT ").append(kCols).append( + " FROM pairings WHERE token_sha256 = ?1 AND revoked_at IS NULL")); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, std::string_view(hash)); !r) return std::unexpected(r.error()); + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return std::optional{}; + return std::optional{read_row(*st)}; +} + +DbResult Pairings::touch(std::string_view pairing_id, std::string_view now_iso) { + auto st = db_.prepare("UPDATE pairings SET last_seen_at = ?2 WHERE pairing_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, pairing_id); !r) return std::unexpected(r.error()); + if (auto r = st->bind(2, now_iso); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return {}; +} + +DbResult Pairings::revoke(std::string_view pairing_id, std::string_view now_iso) { + auto st = db_.prepare( + "UPDATE pairings SET revoked_at = ?2 WHERE pairing_id = ?1 AND revoked_at IS NULL"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, pairing_id); !r) return std::unexpected(r.error()); + if (auto r = st->bind(2, now_iso); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + +DbResult> Pairings::list_active() { + auto st = db_.prepare(std::string("SELECT ").append(kCols).append( + " FROM pairings WHERE revoked_at IS NULL ORDER BY created_at")); + if (!st) return std::unexpected(st.error()); + std::vector out; + for (;;) { + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + out.push_back(read_row(*st)); + } + return out; +} + +} // namespace velox::daemon::store diff --git a/daemon/src/store/pairings.hpp b/daemon/src/store/pairings.hpp new file mode 100644 index 0000000..ad95ad5 --- /dev/null +++ b/daemon/src/store/pairings.hpp @@ -0,0 +1,53 @@ +#pragma once + +// Access to the `pairings` table: the WebSocket transport's revocable per-install tokens +// (docs/05 §4). The plaintext token is returned by create() exactly once and never +// stored — only its SHA-256 (CLAUDE.md §4). + +#include +#include +#include +#include + +#include "store/sqlite.hpp" + +namespace velox::daemon::store { + +struct Pairing { + std::string pairing_id; + std::string origin; + std::string label; + std::string created_at; + std::optional last_seen_at; + std::optional revoked_at; +}; + +class Pairings { +public: + explicit Pairings(Db& db) : db_(db) {} + + struct Created { + std::string pairing_id; + std::string token; // plaintext — send once, to the client, then forget + }; + + // Mint a token for `origin`, store its hash + `label`, timestamp `now_iso`. + DbResult create(std::string_view origin, std::string_view label, + std::string_view now_iso); + + // The active (non-revoked) pairing whose token hashes to this value, if any. + DbResult> find_active_by_token(std::string_view plaintext_token); + + // Bump last_seen_at. Called on every authenticated connect. + DbResult touch(std::string_view pairing_id, std::string_view now_iso); + + // Mark revoked. Returns false if there was no such active pairing. + DbResult revoke(std::string_view pairing_id, std::string_view now_iso); + + DbResult> list_active(); + +private: + Db& db_; +}; + +} // namespace velox::daemon::store diff --git a/daemon/src/store/sqlite.cpp b/daemon/src/store/sqlite.cpp index 136d60e..1dd1726 100644 --- a/daemon/src/store/sqlite.cpp +++ b/daemon/src/store/sqlite.cpp @@ -2,6 +2,8 @@ #include +#include + #include namespace velox::daemon::store { @@ -38,6 +40,13 @@ DbResult Db::open(const std::string& path) { } Db db(handle); + // Not a secret store (credentials go to the Secret Service), but task URLs and pairing + // hashes still are not world-readable. SQLite honours the umask; pin 0600 explicitly. + if (path != ":memory:" && !path.empty() && path.front() != ':') { + ::chmod(path.c_str(), 0600); + ::chmod((path + "-wal").c_str(), 0600); + ::chmod((path + "-shm").c_str(), 0600); + } // WAL for crash-safe concurrent readers (docs/01 §1). busy_timeout so a writer waits // rather than returning SQLITE_BUSY under the RPC loop. foreign_keys is per-connection. for (const char* pragma : {"PRAGMA journal_mode=WAL", "PRAGMA synchronous=NORMAL", diff --git a/daemon/src/util/crypto.cpp b/daemon/src/util/crypto.cpp new file mode 100644 index 0000000..39b1b7c --- /dev/null +++ b/daemon/src/util/crypto.cpp @@ -0,0 +1,57 @@ +#include "util/crypto.hpp" + +#include +#include +#include + +#include + +namespace velox::daemon::crypto { + +std::array sha1(std::string_view data) { + std::array out{}; + ::SHA1(reinterpret_cast(data.data()), data.size(), out.data()); + return out; +} + +std::string sha256_hex(std::string_view data) { + unsigned char digest[SHA256_DIGEST_LENGTH]; + ::SHA256(reinterpret_cast(data.data()), data.size(), digest); + + static constexpr char kHex[] = "0123456789abcdef"; + std::string out; + out.reserve(SHA256_DIGEST_LENGTH * 2); + for (unsigned char b : digest) { + out.push_back(kHex[b >> 4]); + out.push_back(kHex[b & 0x0F]); + } + return out; +} + +std::string base64_encode(const std::uint8_t* data, std::size_t len) { + // 4 chars per 3 bytes, rounded up, plus a NUL that EVP writes. + std::string out(4 * ((len + 2) / 3), '\0'); + const int n = ::EVP_EncodeBlock(reinterpret_cast(out.data()), data, + static_cast(len)); + if (n < 0) throw std::runtime_error("EVP_EncodeBlock failed"); + out.resize(static_cast(n)); + return out; +} + +std::string random_token(std::size_t n) { + std::string raw(n, '\0'); + if (::RAND_bytes(reinterpret_cast(raw.data()), static_cast(n)) != 1) { + throw std::runtime_error("RAND_bytes failed"); + } + std::string b64 = + base64_encode(reinterpret_cast(raw.data()), raw.size()); + // base64 -> base64url, and drop '=' padding. + for (char& c : b64) { + if (c == '+') c = '-'; + else if (c == '/') c = '_'; + } + while (!b64.empty() && b64.back() == '=') b64.pop_back(); + return b64; +} + +} // namespace velox::daemon::crypto diff --git a/daemon/src/util/crypto.hpp b/daemon/src/util/crypto.hpp new file mode 100644 index 0000000..0d31ec1 --- /dev/null +++ b/daemon/src/util/crypto.hpp @@ -0,0 +1,29 @@ +#pragma once + +// Small cryptographic helpers over libcrypto: the WebSocket accept-key hash, the pairing +// token hash, base64, and a CSPRNG token. Nothing bespoke — thin wrappers so callers do +// not touch the OpenSSL API directly. + +#include +#include +#include +#include +#include + +namespace velox::daemon::crypto { + +// SHA-1 of `data`, raw 20 bytes. Used only for the RFC 6455 Sec-WebSocket-Accept value. +std::array sha1(std::string_view data); + +// SHA-256 of `data` as lowercase hex (64 chars). The pairings table stores this, never +// the token itself (CLAUDE.md §4). +std::string sha256_hex(std::string_view data); + +// Standard base64 (with '+' '/' '='), used for Sec-WebSocket-Accept. +std::string base64_encode(const std::uint8_t* data, std::size_t len); + +// `n` bytes from the system CSPRNG, encoded base64url without padding. The pairing token +// is 32 bytes -> 43 chars, matching SessionPairResult.token's "256 bits, base64url". +std::string random_token(std::size_t n = 32); + +} // namespace velox::daemon::crypto diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index fa8ee78..cd7269f 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -1,19 +1,18 @@ # daemon unit + integration tests. Registered with ctest; run via `ctest --preset dev`. -# No external test framework — each file is a small self-checking binary (matches the -# lightweight style core/ uses, without depending on core's private test support). +# No external test framework — each file is a small self-checking binary. -add_executable(veloxd_ndjson_test ndjson_test.cpp) -target_link_libraries(veloxd_ndjson_test PRIVATE veloxd_rpc) -target_compile_options(veloxd_ndjson_test PRIVATE -Wall -Wextra -Wpedantic -Werror) -add_test(NAME veloxd.ndjson COMMAND veloxd_ndjson_test) +function(veloxd_test name) + cmake_parse_arguments(T "" "" "LIBS" ${ARGN}) + add_executable(veloxd_${name}_test ${name}_test.cpp) + target_link_libraries(veloxd_${name}_test PRIVATE ${T_LIBS}) + target_compile_options(veloxd_${name}_test PRIVATE -Wall -Wextra -Wpedantic -Werror) + add_test(NAME veloxd.${name} COMMAND veloxd_${name}_test) + set_tests_properties(veloxd.${name} PROPERTIES TIMEOUT 30) +endfunction() -add_executable(veloxd_uds_roundtrip_test uds_roundtrip_test.cpp) -target_link_libraries(veloxd_uds_roundtrip_test PRIVATE veloxd_rpc) -target_compile_options(veloxd_uds_roundtrip_test PRIVATE -Wall -Wextra -Wpedantic -Werror) -add_test(NAME veloxd.uds_roundtrip COMMAND veloxd_uds_roundtrip_test) -set_tests_properties(veloxd.uds_roundtrip PROPERTIES TIMEOUT 30) - -add_executable(veloxd_store_migrations_test store_migrations_test.cpp) -target_link_libraries(veloxd_store_migrations_test PRIVATE veloxd_store) -target_compile_options(veloxd_store_migrations_test PRIVATE -Wall -Wextra -Wpedantic -Werror) -add_test(NAME veloxd.store_migrations COMMAND veloxd_store_migrations_test) +veloxd_test(ndjson LIBS veloxd_rpc) +veloxd_test(uds_roundtrip LIBS veloxd_rpc) +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) diff --git a/daemon/tests/pairings_test.cpp b/daemon/tests/pairings_test.cpp new file mode 100644 index 0000000..2028e71 --- /dev/null +++ b/daemon/tests/pairings_test.cpp @@ -0,0 +1,87 @@ +// The pairings table + the pairing rate limiter. + +#include + +#include "check.hpp" +#include "rpc/pairing.hpp" +#include "store/migrations.hpp" +#include "store/pairings.hpp" +#include "store/sqlite.hpp" + +using namespace velox::daemon; + +void run() { + // --- store: create -> find-by-token -> revoke --------------------------------- + { + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + + store::Pairings p(*db); + auto created = p.create("moz-extension://abc", "Velox for Firefox", "2026-09-10T00:00:00Z"); + CHECK(created.has_value()); + if (!created) return; + CHECK(created->token.size() >= 40); // 32 bytes base64url, unpadded + CHECK(!created->pairing_id.empty()); + + // The plaintext token is not in the DB — only its hash. + auto st = db->prepare("SELECT count(*) FROM pairings WHERE token_sha256 = ?1"); + CHECK(st.has_value()); + CHECK(st->bind(1, std::string_view(created->token)).has_value()); + auto row = st->step(); + CHECK(row.has_value() && *row); + CHECK_EQ(st->column_int(0), 0); // token itself never stored + + auto found = p.find_active_by_token(created->token); + CHECK(found.has_value()); + CHECK(found->has_value()); + if (found && *found) CHECK_EQ((*found)->origin, std::string("moz-extension://abc")); + + auto missing = p.find_active_by_token("not-the-token"); + CHECK(missing.has_value() && !missing->has_value()); + + auto revoked = p.revoke(created->pairing_id, "2026-09-10T01:00:00Z"); + CHECK(revoked.has_value() && *revoked == true); + + auto after = p.find_active_by_token(created->token); + CHECK(after.has_value() && !after->has_value()); // revoked -> not active + + auto revoke_again = p.revoke(created->pairing_id, "2026-09-10T02:00:00Z"); + CHECK(revoke_again.has_value() && *revoke_again == false); + } + + // --- rate limiter: 5 failures, then a lockout with retryAfter ----------------- + { + rpc::PairingRateLimiter rl; + using Clock = rpc::PairingRateLimiter::Clock; + const auto t0 = Clock::now(); + + for (int i = 0; i < 5; ++i) { + CHECK(rl.check("origin-a", t0).allowed); + rl.record_failure("origin-a", t0); + } + const auto d = rl.check("origin-a", t0); + CHECK(!d.allowed); + CHECK(d.retry_after_sec > 0 && d.retry_after_sec <= 61); + + // A different origin is unaffected — the lockout is per-origin. + CHECK(rl.check("origin-b", t0).allowed); + + // Still locked 30 s later; clear after the lockout elapses. + CHECK(!rl.check("origin-a", t0 + std::chrono::seconds(30)).allowed); + CHECK(rl.check("origin-a", t0 + std::chrono::seconds(121)).allowed); + + // A success wipes the origin's history. + rl.record_failure("origin-c", t0); + rl.record_failure("origin-c", t0); + rl.record_success("origin-c"); + for (int i = 0; i < 4; ++i) { + CHECK(rl.check("origin-c", t0).allowed); + rl.record_failure("origin-c", t0); + } + CHECK(rl.check("origin-c", t0).allowed); // only 4 since the reset + } +} + +TEST_MAIN() diff --git a/daemon/tests/ws_frame_test.cpp b/daemon/tests/ws_frame_test.cpp new file mode 100644 index 0000000..a3283b8 --- /dev/null +++ b/daemon/tests/ws_frame_test.cpp @@ -0,0 +1,150 @@ +#include "rpc/ws_frame.hpp" +#include "rpc/ws_handshake.hpp" + +#include +#include + +#include "check.hpp" + +using namespace velox::daemon::rpc; + +namespace { + +// Build a *client* frame: FIN/opcode, mask bit set, a fixed 4-byte mask, masked payload. +std::string client_frame(WsOpcode op, std::string_view payload, bool fin = true) { + std::string f; + f.push_back(static_cast((fin ? 0x80 : 0x00) | static_cast(op))); + const std::size_t n = payload.size(); + if (n < 126) { + f.push_back(static_cast(0x80 | n)); + } else if (n <= 0xFFFF) { + f.push_back(static_cast(0x80 | 126)); + f.push_back(static_cast((n >> 8) & 0xFF)); + f.push_back(static_cast(n & 0xFF)); + } else { + f.push_back(static_cast(0x80 | 127)); + for (int i = 7; i >= 0; --i) + f.push_back(static_cast((static_cast(n) >> (i * 8)) & 0xFF)); + } + const char key[4] = {0x12, 0x34, 0x56, 0x78}; + f.append(key, 4); + for (std::size_t i = 0; i < n; ++i) f.push_back(static_cast(payload[i] ^ key[i & 3])); + return f; +} + +} // namespace + +void run() { + // --- one text frame ------------------------------------------------------------ + { + WsFrameReader r; + std::vector m; + CHECK(r.feed(client_frame(WsOpcode::Text, "{\"a\":1}"), m) == WsFrameReader::Status::Ok); + CHECK_EQ(m.size(), 1u); + CHECK(m[0].opcode == WsOpcode::Text); + CHECK_EQ(m[0].payload, std::string("{\"a\":1}")); + } + + // --- fragmented: text (fin=0) + continuation (fin=1) -------------------------- + { + WsFrameReader r; + std::vector m; + r.feed(client_frame(WsOpcode::Text, "hel", /*fin=*/false), m); + CHECK_EQ(m.size(), 0u); + r.feed(client_frame(WsOpcode::Continuation, "lo", /*fin=*/true), m); + CHECK_EQ(m.size(), 1u); + CHECK_EQ(m[0].payload, std::string("hello")); + } + + // --- byte-at-a-time delivery still reassembles ------------------------------- + { + WsFrameReader r; + std::vector m; + const std::string frame = client_frame(WsOpcode::Text, "streamed"); + for (char ch : frame) r.feed(std::string_view(&ch, 1), m); + CHECK_EQ(m.size(), 1u); + CHECK_EQ(m[0].payload, std::string("streamed")); + } + + // --- a 200-byte payload exercises the 16-bit length path -------------------- + { + WsFrameReader r; + std::vector m; + const std::string big(200, 'x'); + r.feed(client_frame(WsOpcode::Text, big), m); + CHECK_EQ(m.size(), 1u); + CHECK_EQ(m[0].payload.size(), 200u); + } + + // --- ping is surfaced so the server can pong ------------------------------- + { + WsFrameReader r; + std::vector m; + r.feed(client_frame(WsOpcode::Ping, "hi"), m); + CHECK_EQ(m.size(), 1u); + CHECK(m[0].opcode == WsOpcode::Ping); + } + + // --- an unmasked client frame is a protocol error (RFC 6455 §5.1) ---------- + { + WsFrameReader r; + std::vector m; + std::string bad; + bad.push_back(static_cast(0x81)); // FIN + text + bad.push_back(static_cast(0x03)); // len 3, mask bit clear + bad.append("abc"); + CHECK(r.feed(bad, m) == WsFrameReader::Status::ProtocolError); + } + + // --- a declared length past the cap is rejected before allocating ---------- + { + WsFrameReader r; + std::vector m; + std::string hdr; + hdr.push_back(static_cast(0x82)); // FIN + binary + hdr.push_back(static_cast(0x80 | 127)); + for (int i = 7; i >= 0; --i) + hdr.push_back(static_cast((0x0000000001000000ull >> (i * 8)) & 0xFF)); // 16 MiB + CHECK(r.feed(hdr, m) == WsFrameReader::Status::MessageTooBig); + } + + // --- ws_encode: server frames are unmasked, correct length byte ------------ + { + const std::string f = ws_encode(WsOpcode::Text, "abc"); + CHECK_EQ(static_cast(f[0]), 0x81u); + CHECK_EQ(static_cast(f[1]), 0x03u); // len 3, no mask bit + CHECK_EQ(f.substr(2), std::string("abc")); + } + + // --- RFC 6455 §1.3 sample accept value ------------------------------------ + CHECK_EQ(ws_accept_key("dGhlIHNhbXBsZSBub25jZQ=="), + std::string("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=")); + + // --- handshake: a page origin is refused, an extension origin upgrades ----- + { + const std::string req_page = + "GET / HTTP/1.1\r\nHost: 127.0.0.1:52000\r\nUpgrade: websocket\r\n" + "Connection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + "Sec-WebSocket-Version: 13\r\nOrigin: https://evil.example\r\n\r\n"; + const auto r = ws_try_handshake(req_page); + CHECK(r.complete); + CHECK(!r.ok); + CHECK(r.response.find("403") != std::string::npos); + } + { + const std::string req_ext = + "GET / HTTP/1.1\r\nHost: 127.0.0.1:52000\r\nUpgrade: websocket\r\n" + "Connection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + "Sec-WebSocket-Version: 13\r\n" + "Origin: moz-extension://11111111-2222-3333-4444-555555555555\r\n\r\n"; + const auto r = ws_try_handshake(req_ext); + CHECK(r.complete); + CHECK(r.ok); + CHECK(r.response.find("101") != std::string::npos); + CHECK(r.response.find("Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=") != + std::string::npos); + CHECK_EQ(r.origin, std::string("moz-extension://11111111-2222-3333-4444-555555555555")); + } +} + +TEST_MAIN() diff --git a/daemon/tests/ws_server_test.cpp b/daemon/tests/ws_server_test.cpp new file mode 100644 index 0000000..e241012 --- /dev/null +++ b/daemon/tests/ws_server_test.cpp @@ -0,0 +1,216 @@ +// Integration: a real WsServer on a loopback port, a hand-rolled WebSocket client. +// Covers the handshake, the pairing flow, the token gate (-32002), and the +// privileged-over-WS refusal (-32003). + +#include +#include +#include +#include + +#include +#include +#include +#include + +#include + +#include "check.hpp" +#include "rpc/dispatcher.hpp" +#include "rpc/event_loop.hpp" +#include "rpc/pairing.hpp" +#include "rpc/runtime_dir.hpp" +#include "rpc/ws_frame.hpp" +#include "rpc/ws_server.hpp" +#include "store/migrations.hpp" +#include "store/sqlite.hpp" + +using nlohmann::json; +namespace rpc = velox::daemon::rpc; +namespace store = velox::daemon::store; + +namespace { + +int dial(int port) { + const int fd = ::socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); + sockaddr_in a{}; + a.sin_family = AF_INET; + a.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK); + a.sin_port = ::htons(static_cast(port)); + if (::connect(fd, reinterpret_cast(&a), sizeof(a)) != 0) { + ::close(fd); + return -1; + } + return fd; +} + +void write_all(int fd, std::string_view s) { + std::size_t off = 0; + while (off < s.size()) { + const ssize_t n = ::write(fd, s.data() + off, s.size() - off); + if (n <= 0) return; + off += static_cast(n); + } +} + +std::string read_some(int fd) { + char buf[8192]; + const ssize_t n = ::read(fd, buf, sizeof(buf)); + return n > 0 ? std::string(buf, static_cast(n)) : std::string{}; +} + +// A masked client text frame. +std::string client_text(std::string_view payload) { + std::string f; + f.push_back(static_cast(0x81)); // FIN + text + const std::size_t n = payload.size(); + if (n < 126) { + f.push_back(static_cast(0x80 | n)); + } else { + f.push_back(static_cast(0x80 | 126)); + f.push_back(static_cast((n >> 8) & 0xFF)); + f.push_back(static_cast(n & 0xFF)); + } + const char k[4] = {0x0A, 0x0B, 0x0C, 0x0D}; + f.append(k, 4); + for (std::size_t i = 0; i < n; ++i) f.push_back(static_cast(payload[i] ^ k[i & 3])); + return f; +} + +// Decode one unmasked server frame from `buf`, consuming it. Returns payload; sets `op`. +std::string server_frame(std::string& buf, rpc::WsOpcode& op) { + if (buf.size() < 2) return {}; + op = static_cast(buf[0] & 0x0F); + std::size_t len = static_cast(buf[1]) & 0x7F; + std::size_t header = 2; + if (len == 126) { + len = (static_cast(static_cast(buf[2])) << 8) | + static_cast(buf[3]); + header = 4; + } + if (buf.size() < header + len) return {}; + std::string payload = buf.substr(header, len); + buf.erase(0, header + len); + return payload; +} + +// Send a request frame, wait for one text reply, return its parsed JSON. +json rpc_call(int fd, const json& req) { + write_all(fd, client_text(req.dump())); + std::string buf; + for (;;) { + buf += read_some(fd); + rpc::WsOpcode op{}; + std::string save = buf; + const std::string payload = server_frame(buf, op); + if (payload.empty() && buf == save) continue; // need more bytes + if (op == rpc::WsOpcode::Text) return json::parse(payload, nullptr, false); + } +} + +} // namespace + +void run() { + ::unsetenv("VELOX_PAIR_AUTO"); + + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + + char tmpl[] = "/tmp/velox-ws-test-XXXXXX"; + const char* dir = ::mkdtemp(tmpl); + CHECK(dir != nullptr); + rpc::RuntimeDir rt{dir ? dir : "/tmp"}; + + rpc::EventLoop loop; + rpc::VeloxDispatcher dispatcher; + rpc::EnvAutoApprover approver; + rpc::WsServer server(loop, dispatcher, *db, approver, rt); + const auto ec = server.start(); + CHECK(!ec); + if (ec) return; + CHECK(server.port() >= rpc::WsServer::kPortLo); + CHECK(server.port() <= rpc::WsServer::kPortHi); + + std::thread th([&loop] { loop.run(); }); + + const std::string origin = "moz-extension://11111111-2222-3333-4444-555555555555"; + + // --- handshake --------------------------------------------------------------- + const int fd = dial(server.port()); + CHECK(fd >= 0); + write_all(fd, + "GET / HTTP/1.1\r\nHost: 127.0.0.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n" + "Origin: " + origin + "\r\n\r\n"); + std::string hs; + while (hs.find("\r\n\r\n") == std::string::npos) hs += read_some(fd); + CHECK(hs.find("101 Switching Protocols") != std::string::npos); + + // --- session.hello with no token -> -32002 -------------------------------- + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 1}, {"method", "session.hello"}, + {"params", {{"clientType", "extension"}, + {"clientName", "Velox for Firefox"}, + {"protocolVersion", + std::string(velox::proto::kProtocolVersion)}}}}); + CHECK(r.contains("error")); + CHECK_EQ(r["error"]["code"].get(), -32002); + } + + // --- session.pair with approval off -> not approved ---------------------- + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 2}, {"method", "session.pair"}, + {"params", {{"clientName", "Velox for Firefox"}, + {"extensionId", "11111111-2222-3333-4444-555555555555"}}}}); + CHECK(r.contains("error")); + } + + // --- approval on -> a token, then hello with it succeeds ---------------- + ::setenv("VELOX_PAIR_AUTO", "1", 1); + std::string token; + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 3}, {"method", "session.pair"}, + {"params", {{"clientName", "Velox for Firefox"}, + {"extensionId", "11111111-2222-3333-4444-555555555555"}}}}); + CHECK(r.contains("result")); + if (r.contains("result")) { + token = r["result"]["token"].get(); + CHECK(token.size() >= 40); + } + } + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 4}, {"method", "session.hello"}, + {"params", {{"clientType", "extension"}, + {"clientName", "Velox for Firefox"}, + {"protocolVersion", + std::string(velox::proto::kProtocolVersion)}, + {"token", token}}}}); + CHECK(r.contains("result")); + if (r.contains("result")) + CHECK_EQ(r["result"]["transport"].get(), std::string("ws")); + } + + // --- privileged method over WS -> -32003 ------------------------------- + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 5}, {"method", "settings.get"}, + {"params", {{"keys", nullptr}}}}); + CHECK(r.contains("error")); + CHECK_EQ(r["error"]["code"].get(), -32003); + } + + // --- a non-privileged method while authed -> a real result ----------- + { + const json r = rpc_call(fd, {{"jsonrpc", "2.0"}, {"id", 6}, {"method", "download.list"}, + {"params", json::object()}}); + CHECK(r.contains("result")); + if (r.contains("result")) CHECK_EQ(r["result"]["total"].get(), 0); + } + + ::close(fd); + loop.stop(); + th.join(); + ::unlink(rt.ws_port_path().c_str()); +} + +TEST_MAIN()