trevon/glove
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
glove/tests/control/receipt_audit_unix_server_test.cpp
0xkite 5c48e5fd8a
 fix(control): report start replay disposition, hard-cap jsonl audit journal, deterministic control tests 
MEDIUM 1: session_runtime::start() now returns an explicit
fresh-launch vs replay disposition (session_start_result), and
handle_start_session sets outcome.applied only for a fresh committed
launch. An authenticated idempotent start replay therefore never gains
connection-scoped teardown authority: a broken pipe during a replay can
no longer stop an already-running guest. The control audit event for a
replayed delivery carries the replay disposition
(start_session:replay:<session>).

MEDIUM 2: jsonl_sink truncation now accounts for record delimiters
exactly, so current_bytes_ always tracks the physical file; only
newline-terminated complete records are retained (no blank or partial
lines, bounded read window); the cap is a hard cap (an oversized event
is dropped with a structured operator note; a pre-existing oversized
journal is bounded with a bounded read at construction).

MEDIUM 3: the broken-pipe degrade test now synchronizes through a
mutex-protected condition-variable gate handshake that proves the
dispatch is parked inside the blocked gate before the client resets,
records the start only after the dispatch completes, and the
mark_stopping test destroys every registry lock holder (runtime,
protocol, registry) before the independent registry reopen so the
nonblocking exclusive flock can actually be acquired.
2026-08-31 09:34:45 -06:00

1291 lines
52 KiB
C++
Raw Permalink Blame History

#include "glove/audit/sink.hpp"
#include "glove/container/receipt_producer.hpp"
#include "glove/control/receipt_audit_protocol.hpp"
#include "glove/control/receipt_audit_unix_server.hpp"
#include "glove/supervisor/path_alias.hpp"
#include "glove/supervisor/session_plan.hpp"
#include "receipt_audit_unix_server_detail.hpp"
#include "receipt_audit_wire.hpp"
#include <arpa/inet.h>
#include <glaze/glaze.hpp>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <unistd.h>
#include <cerrno>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <limits>
#include <mutex>
#include <optional>
#include <set>
#include <string>
#include <string_view>
#include <system_error>
#include <thread>
#include <utility>
#include <vector>
namespace wire_test {
struct rpc_error {
std::string code;
std::string message;
};
struct rpc_response {
std::string jsonrpc;
std::string id;
std::optional<glz::raw_json> result;
std::optional<rpc_error> error;
};
struct page_result {
std::uint8_t schema_version = 0;
std::vector<glove::container::authenticated_resource_enforcement_receipt> envelopes;
std::vector<glove::container::authenticated_refinement_evaluation_receipt> refinement_envelopes;
bool has_more = false;
glove::container::receipt_audit_anchor local_anchor;
};
} // namespace wire_test
namespace {
#define REQUIRE(cond) \
do { \
if (!(cond)) { \
std::fprintf(stderr, "REQUIRE failed: %s @ %s:%d\n", #cond, __FILE__, __LINE__); \
return 1; \
} \
} while (0)
constexpr std::string_view audit_key =
"000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
constexpr std::string_view audit_key_id =
"b97d6f8d2ae381761ea00f360c230cf75e8de5fdc6a8d25624a5c36b97f0d475";
constexpr std::string_view bootstrap_secret =
"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
constexpr std::string_view plan_digest =
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
class temporary_directory {
public:
temporary_directory() {
std::string pattern = "/tmp/glove-receipt-server-test-XXXXXX";
if (char* created = ::mkdtemp(pattern.data()); created != nullptr) {
root_ = created;
}
}
temporary_directory(const temporary_directory&) = delete;
auto operator=(const temporary_directory&) -> temporary_directory& = delete;
~temporary_directory() {
std::error_code error;
std::filesystem::remove_all(root_, error);
}
[[nodiscard]] auto root() const -> const std::filesystem::path& { return root_; }
private:
std::filesystem::path root_;
};
class unique_fd {
public:
explicit unique_fd(int value = -1) noexcept : value_{value} {}
unique_fd(const unique_fd&) = delete;
auto operator=(const unique_fd&) -> unique_fd& = delete;
unique_fd(unique_fd&& other) noexcept : value_{std::exchange(other.value_, -1)} {}
auto operator=(unique_fd&&) -> unique_fd& = delete;
~unique_fd() {
if (value_ >= 0) {
::close(value_);
}
}
void reset(int value = -1) noexcept {
if (value_ >= 0) {
::close(value_);
}
value_ = value;
}
[[nodiscard]] auto get() const noexcept -> int { return value_; }
private:
int value_;
};
auto write_owner_only(const std::filesystem::path& path, std::string_view value) -> bool {
std::ofstream output{path, std::ios::binary | std::ios::trunc};
output << value << '\n';
output.flush();
return output.good() && ::chmod(path.c_str(), 0600) == 0;
}
auto receipt() -> glove::container::resource_enforcement_receipt {
using namespace glove::container;
return {
.schema_version = 1,
.profile_digest = std::string(64, 'c'),
.backend = sandbox_backend::linux_production,
.backend_id = "linux-production:cgroup-v2-v1",
.configured_limits =
{
.cpu_time_ms = 60'000,
.memory_bytes = std::uint64_t{512} * 1024U * 1024U,
.pids = 128,
.wall_time_ms = 120'000,
.disk_bytes = std::uint64_t{1024} * 1024U * 1024U,
.terminal_output_bytes = std::uint64_t{16} * 1024U * 1024U,
},
.mechanisms =
{
.cpu_time = enforcement_mechanism::cgroup_v2,
.memory = enforcement_mechanism::cgroup_v2,
.pids = enforcement_mechanism::cgroup_v2,
.wall_time = enforcement_mechanism::watchdog,
.disk = enforcement_mechanism::filesystem_quota,
.terminal_output = enforcement_mechanism::byte_counter,
.receipt_schema_version = 1,
},
.observed =
{
.cpu_time_ms = 500,
.peak_memory_bytes = std::uint64_t{16} * 1024U * 1024U,
.peak_pids = 2,
.wall_time_ms = 750,
.disk_bytes = 4096,
.terminal_output_bytes = 1024,
},
.termination_cause = resource_termination_cause::exited,
.exit_code = 0,
.started_at_ms = 1'000,
.finished_at_ms = 1'750,
.library_projections = {},
.retained_changes = {},
};
}
auto write_exact(int descriptor, const void* input, std::size_t size) -> bool {
const auto* bytes = static_cast<const std::byte*>(input);
std::size_t written = 0;
while (written < size) {
const auto result = ::write(descriptor, bytes + written, size - written);
if (result < 0 && errno == EINTR) {
continue;
}
if (result <= 0) {
return false;
}
written += static_cast<std::size_t>(result);
}
return true;
}
auto read_exact(int descriptor, void* output, std::size_t size) -> bool {
auto* bytes = static_cast<std::byte*>(output);
std::size_t consumed = 0;
while (consumed < size) {
const auto result = ::read(descriptor, bytes + consumed, size - consumed);
if (result < 0 && errno == EINTR) {
continue;
}
if (result <= 0) {
return false;
}
consumed += static_cast<std::size_t>(result);
}
return true;
}
auto connect_to(const std::filesystem::path& socket_path) -> unique_fd {
unique_fd descriptor{::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0)};
if (descriptor.get() < 0) {
return unique_fd{};
}
::sockaddr_un address{};
address.sun_family = AF_UNIX;
const auto value = socket_path.string();
if (value.size() >= sizeof(address.sun_path)) {
return unique_fd{};
}
std::memcpy(address.sun_path, value.c_str(), value.size() + 1U);
if (::connect(
descriptor.get(), reinterpret_cast<const ::sockaddr*>(&address), sizeof(address)
) != 0) {
return unique_fd{};
}
return descriptor;
}
auto transact(const std::filesystem::path& socket_path, std::string_view frame)
-> std::optional<std::string> {
auto descriptor = connect_to(socket_path);
if (descriptor.get() < 0) {
return std::nullopt;
}
const auto size = htonl(static_cast<std::uint32_t>(frame.size()));
if (!write_exact(descriptor.get(), &size, sizeof(size)) ||
!write_exact(descriptor.get(), frame.data(), frame.size())) {
return std::nullopt;
}
std::uint32_t response_size = 0;
if (!read_exact(descriptor.get(), &response_size, sizeof(response_size))) {
return std::nullopt;
}
const auto decoded_size = ntohl(response_size);
if (decoded_size == 0 || decoded_size > glove::control::max_control_frame_bytes) {
return std::nullopt;
}
std::string response(decoded_size, '\0');
if (!read_exact(descriptor.get(), response.data(), response.size())) {
return std::nullopt;
}
return response;
}
auto make_request(
std::string_view id,
std::string_view method,
std::string_view payload,
std::optional<std::string_view> idempotency_key = std::nullopt,
std::string_view secret = bootstrap_secret,
std::uint64_t deadline_ms = 4102444800000
) -> std::string {
std::string request = "{\"jsonrpc\":\"2.0\",\"id\":\"" + std::string{id} + "\",\"method\":\"" +
std::string{method} +
"\",\"params\":{\"schema_version\":1,\"bootstrap_secret\":\"" +
std::string{secret} + "\",\"deadline_ms\":" + std::to_string(deadline_ms);
if (idempotency_key) {
request += ",\"idempotency_key\":\"" + std::string{*idempotency_key} + "\"";
}
request += ",\"payload\":" + std::string{payload} + "}}";
return request;
}
auto decode_response(std::string_view frame) -> std::optional<wire_test::rpc_response> {
wire_test::rpc_response response;
constexpr glz::opts strict{.error_on_unknown_keys = true};
if (glz::read<strict>(response, frame)) {
return std::nullopt;
}
return response;
}
// Regression stub for connection-scoped delivery degradation: records every
// start() and stop() so tests can prove a guest launched by a genuinely
// authenticated+applied start_session is torn down through the normal
// idempotent stop path when its controller connection fails at the transport
// level, and that unauthenticated or unapplied requests never tear anything
// down.
class degrade_test_runtime final : public glove::control::session_runtime {
public:
degrade_test_runtime() = default;
explicit degrade_test_runtime(bool stop_succeeds) : stop_succeeds_{stop_succeeds} {}
[[nodiscard]] auto backend_id() const noexcept -> std::string_view override {
return "degrade-test";
}
[[nodiscard]] auto agent_runtime_adapter_schema_version() const noexcept
-> std::uint8_t override {
return 0;
}
[[nodiscard]] auto managed_runtime_ids() const -> std::vector<std::string> override {
return {};
}
[[nodiscard]] auto resource_capabilities() const noexcept
-> glove::container::resource_enforcement_capabilities override {
return {};
}
auto start(
glove::container::receipt_audit_producer&,
const glove::control::session_start_authorization& authorization,
std::string_view idempotency_key,
std::uint64_t now_ms
) -> std::expected<glove::control::session_start_result, std::string> override {
// Gate handshake: the block decision is read under the same mutex
// that arms it, and the test observes — via the condition variable —
// that the dispatch is parked inside the blocked gate BEFORE the
// client resets the connection. That makes the broken-pipe outcome
// deterministic instead of racy.
{
std::unique_lock lock{mutex_};
if (block_start_) {
gate_entered_ = true;
start_gate_.notify_all();
start_gate_.wait(lock, [this] { return gate_open_; });
}
}
// The start is recorded only AFTER the dispatch completes (and the
// handler has decided outcome.applied), so cross-thread observers
// never see a start before it is committed.
const bool replay = [this, &idempotency_key] {
const std::scoped_lock lock{mutex_};
return !used_keys_.insert(std::string{idempotency_key}).second;
}();
{
const std::scoped_lock lock{mutex_};
started.emplace_back(authorization.session_id);
}
// State `created` needs no registry projection: handle_start_session
// encodes the record directly. A repeated idempotency key models an
// authenticated idempotent replay: the same record is returned, but
// with the explicit replay disposition (fresh_launch == false), so
// the handler must not treat the replay as applied.
glove::control::session_record record{
.schema_version = 1,
.session_id = authorization.session_id,
.controller_plan_digest = authorization.controller_plan_digest,
.plan_content_digest = authorization.plan_content_digest,
.state = glove::control::session_state::created,
.policy_revision = 1,
.expires_at_ms = authorization.expires_at_ms,
.created_at_ms = now_ms,
};
return glove::control::session_start_result{
.record = std::move(record), .fresh_launch = !replay
};
}
auto reconcile(glove::container::receipt_audit_producer&, std::uint64_t)
-> std::expected<glove::control::session_reconciliation_report, std::string> override {
return glove::control::session_reconciliation_report{};
}
[[nodiscard]] auto list() const
-> std::expected<std::vector<std::string>, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
[[nodiscard]] auto read(std::string_view, std::uint64_t, std::size_t) const
-> std::expected<glove::control::session_transcript_read, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
[[nodiscard]] auto wait_read(std::string_view, std::uint64_t, std::size_t, std::uint64_t)
-> std::expected<glove::control::session_transcript_read, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
auto write_input(std::string_view, std::string_view)
-> std::expected<void, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
auto resize(std::string_view, std::uint16_t, std::uint16_t)
-> std::expected<void, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
auto signal(std::string_view, glove::control::session_signal)
-> std::expected<void, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
auto stop(std::string_view session_id) -> std::expected<void, std::string> override {
const std::scoped_lock lock{mutex_};
stopped.emplace_back(session_id);
if (!stop_succeeds_) {
return std::unexpected(std::string{"stop unavailable"});
}
return {};
}
auto stop(std::string_view session_id, std::string_view)
-> std::expected<void, std::string> override {
return stop(session_id);
}
[[nodiscard]] auto wait(std::string_view)
-> std::expected<glove::control::session_terminal_record, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
auto cleanup(std::string_view) -> std::expected<void, std::string> override {
return std::unexpected(std::string{"unavailable"});
}
// Thread-safe snapshot helpers for cross-thread synchronization.
[[nodiscard]] auto started_snapshot() -> std::vector<std::string> {
const std::scoped_lock lock{mutex_};
return started;
}
[[nodiscard]] auto stopped_snapshot() -> std::vector<std::string> {
const std::scoped_lock lock{mutex_};
return stopped;
}
// Arm the start() gate before the request is sent; the handshake proves
// the dispatch is parked inside the blocked gate before the client
// resets, and release it after the connection is torn down.
void arm_start_gate() {
const std::scoped_lock lock{mutex_};
block_start_ = true;
gate_entered_ = false;
gate_open_ = false;
}
template<class Rep, class Period>
[[nodiscard]] auto wait_for_start_gate(std::chrono::duration<Rep, Period> timeout) -> bool {
std::unique_lock lock{mutex_};
return start_gate_.wait_for(lock, timeout, [this] { return gate_entered_; });
}
void release_start_gate() {
{
const std::scoped_lock lock{mutex_};
gate_open_ = true;
}
start_gate_.notify_all();
}
std::vector<std::string> started;
std::vector<std::string> stopped;
private:
mutable std::mutex mutex_;
std::condition_variable start_gate_;
bool block_start_ = false;
bool gate_entered_ = false;
bool gate_open_ = false;
bool stop_succeeds_ = true;
std::set<std::string> used_keys_;
};
// Minimal bounded plan validator for the full protocol/server stack below
// (mirrors the receipt audit server fixture policy).
auto plan_validator_for(const std::filesystem::path& source)
-> glove::supervisor::result<glove::supervisor::session_plan_validator> {
using namespace glove::supervisor;
auto paths = path_alias_registry::build({
path_alias_policy{
.alias = "sage-protocol",
.host_path = std::filesystem::canonical(source).string(),
.target_path = "/workspace/sage-protocol",
.max_ttl_secs = 600,
.access = {
path_access_policy{
.access = path_access::ephemeral_write,
.materialization = path_materialization::copy,
.create_policy = path_create_policy::empty_directory,
.cleanup_policy = path_cleanup_policy::remove,
.max_bytes = 1'000'000,
},
},
},
});
if (!paths) {
return std::unexpected(paths.error());
}
return session_plan_validator::build(
session_plan_policy{
.revision = 1,
.max_plan_ttl_ms = 180'000,
.runtime_templates =
{
runtime_template_policy{
.runtime_template_id = "codex-headless-v1",
.runtime_id = "codex",
.adapter_command_digest = std::string(64, 'a'),
.backend = sandbox_backend::linux_production,
.allowed_path_aliases = {"sage-protocol"},
.allowed_projection_destinations = {"codex-skills"},
.launch = {},
.adoption = std::nullopt,
},
},
.library_projection_destinations =
{
library_projection_destination_policy{
.alias = "codex-skills",
.target_path = "/opt/sage/codex-skills",
},
},
.resource_profiles =
{
resource_limits{
.cpu_time_ms = 60'000,
.memory_bytes = std::uint64_t{512} * 1024U * 1024U,
.pids = 128,
.wall_time_ms = 120'000,
.disk_bytes = std::uint64_t{1024} * 1024U * 1024U,
.terminal_output_bytes = std::uint64_t{16} * 1024U * 1024U,
},
},
.egress_policy_ids = {"deny-all"},
.tool_policy_ids = {"no-upstream-tools"},
.secret_handles = {"github-readonly"},
.egress_policies = {},
.secret_mounts = {},
},
std::move(*paths)
);
}
// Regression test (Workflow #218, sol-audit MEDIUM): degradation authority is
// bound to the REAL authenticated/applied dispatch outcome. handle_frame must
// produce structured metadata, and degrade_failed_delivery must act ONLY on
// genuinely authenticated+applied start_session frames — never by re-decoding
// a raw frame. Wrong-secret, expired-deadline, rejected-start, and non-start
// outcomes must never stop a guest; they record only the connection-scoped
// audit event.
auto verify_degrade_requires_authenticated_applied_outcome() -> int {
temporary_directory temp;
REQUIRE(!temp.root().empty());
const auto audit_key_path = temp.root() / "audit.key";
const auto journal_path = temp.root() / "receipts.journal";
const auto sessions_path = temp.root() / "sessions.journal";
const auto plan_source = temp.root() / "plan-source";
REQUIRE(std::filesystem::create_directory(plan_source));
std::ofstream{plan_source / "tracked.txt"} << "host-owned\n";
REQUIRE(write_owner_only(audit_key_path, audit_key));
auto validator_result = plan_validator_for(plan_source);
REQUIRE(validator_result.has_value());
auto validator = std::make_shared<const glove::supervisor::session_plan_validator>(
std::move(*validator_result)
);
auto sessions = glove::control::session_registry::open_or_create(sessions_path, validator);
REQUIRE(sessions.has_value());
auto registry = std::shared_ptr<glove::control::session_registry>{std::move(*sessions)};
const glove::container::receipt_audit_producer_config producer_config{
.key_path = audit_key_path,
.journal_path = journal_path,
};
// Initialize + acknowledge so the real dispatch (receipt paging) can
// bootstrap a reconciled producer from the journal.
auto seed_producer = glove::container::receipt_audit_producer::initialize(producer_config);
REQUIRE(seed_producer.has_value());
REQUIRE((*seed_producer)->acknowledge_bootstrap((*seed_producer)->anchor()).has_value());
seed_producer->reset();
auto audit_sink = glove::audit::make_memory_sink();
auto runtime = std::make_shared<degrade_test_runtime>();
const glove::control::receipt_audit_unix_server_config config{
.socket_path = {},
.bootstrap_secret_path = {},
.producer = producer_config,
.plan_validator = validator,
.sessions = registry,
.runtime = runtime,
.local_services = {},
.path_exposures = {},
.materialization_root = (temp.root() / "materializations").string(),
.io_timeout_ms = 5'000,
.control_audit = audit_sink,
};
auto protocol = glove::control::receipt_audit_protocol::create(
bootstrap_secret,
producer_config,
validator,
registry,
runtime,
{},
config.materialization_root
);
REQUIRE(protocol.has_value());
const auto start_payload =
R"({"authorization":{"schema_version":1,"authorization_id":"auth-1",)"
R"("session_id":"session-9","controller_plan_digest":")" +
std::string(plan_digest) + R"(","plan_content_digest":")" + std::string(plan_digest) +
R"(","approved_at_ms":1,"expires_at_ms":4102444800000}})";
// Initialize the producer through the real dispatch (receipt paging).
const glove::container::receipt_audit_anchor genesis{
.key_id = std::string{audit_key_id},
.sequence = 0,
.head_hmac = std::string(64, '0'),
};
auto genesis_json = glz::write_json(genesis);
REQUIRE(genesis_json.has_value());
auto page_frame = (*protocol)->handle_frame(
make_request(
"page-1", "verify_audit_chain", "{\"sage_anchor\":" + *genesis_json + ",\"limit\":10}"
),
1'000
);
REQUIRE(page_frame.has_value());
// A genuinely authenticated+applied start_session: the outcome metadata
// must say so, and degradation must tear the guest down.
glove::control::receipt_control_outcome outcome;
auto start_frame = (*protocol)->handle_frame(
make_request("start-1", "start_session", start_payload, "degrade-start-1"), 1'000, &outcome
);
REQUIRE(start_frame.has_value());
auto start_response = decode_response(*start_frame);
REQUIRE(start_response.has_value());
REQUIRE(start_response->result.has_value());
REQUIRE(outcome.authenticated);
REQUIRE(outcome.applied);
REQUIRE(outcome.response_success);
REQUIRE(outcome.method == "start_session");
REQUIRE(outcome.session_id == "session-9");
REQUIRE(runtime->started_snapshot() == std::vector<std::string>{"session-9"});
glove::control::detail::degrade_failed_delivery(
config, outcome, "write control frame: Broken pipe"
);
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// An authenticated idempotent start replay (same request, same
// idempotency key): the response is still the existing record, but the
// outcome must mark it as a replay — never applied — so a broken
// delivery during a replay must NOT tear down the already-running
// guest.
glove::control::receipt_control_outcome replay_outcome;
auto replay_frame = (*protocol)->handle_frame(
make_request("start-1-replay", "start_session", start_payload, "degrade-start-1"),
1'000,
&replay_outcome
);
REQUIRE(replay_frame.has_value());
auto replay_response = decode_response(*replay_frame);
REQUIRE(replay_response.has_value());
REQUIRE(replay_response->result.has_value());
REQUIRE(replay_response->result->str == start_response->result->str);
REQUIRE(replay_outcome.authenticated);
REQUIRE(replay_outcome.replay);
REQUIRE(!replay_outcome.applied);
glove::control::detail::degrade_failed_delivery(
config, replay_outcome, "write control frame: Broken pipe"
);
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// Wrong secret: unauthenticated. No stop attempt; only the audit event.
glove::control::receipt_control_outcome bad_secret_outcome;
auto bad_secret_frame = (*protocol)->handle_frame(
make_request(
"start-2",
"start_session",
start_payload,
"degrade-start-2",
"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
),
1'000,
&bad_secret_outcome
);
REQUIRE(bad_secret_frame.has_value());
auto bad_secret_response = decode_response(*bad_secret_frame);
REQUIRE(bad_secret_response.has_value());
REQUIRE(bad_secret_response->error.has_value());
REQUIRE(bad_secret_response->error->code == "unauthorized");
REQUIRE(!bad_secret_outcome.authenticated);
REQUIRE(!bad_secret_outcome.applied);
REQUIRE(bad_secret_outcome.method == "start_session");
REQUIRE(bad_secret_outcome.session_id.empty());
glove::control::detail::degrade_failed_delivery(
config, bad_secret_outcome, "write control frame: Broken pipe"
);
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// Expired deadline: unauthenticated for degradation purposes.
glove::control::receipt_control_outcome expired_outcome;
auto expired_frame = (*protocol)->handle_frame(
make_request(
"start-3", "start_session", start_payload, "degrade-start-3", bootstrap_secret, 1
),
2'000,
&expired_outcome
);
REQUIRE(expired_frame.has_value());
auto expired_response = decode_response(*expired_frame);
REQUIRE(expired_response.has_value());
REQUIRE(expired_response->error.has_value());
REQUIRE(expired_response->error->code == "deadline_exceeded");
REQUIRE(!expired_outcome.authenticated);
REQUIRE(!expired_outcome.applied);
glove::control::detail::degrade_failed_delivery(config, expired_outcome, "deadline");
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// Non-start methods never gain teardown authority.
glove::control::receipt_control_outcome health_outcome;
auto health_frame = (*protocol)->handle_frame(
make_request("health-1", "health", "null"), 1'000, &health_outcome
);
REQUIRE(health_frame.has_value());
REQUIRE(health_outcome.authenticated);
REQUIRE(health_outcome.method == "health");
REQUIRE(health_outcome.session_id.empty());
REQUIRE(!health_outcome.applied);
glove::control::detail::degrade_failed_delivery(config, health_outcome, "timeout");
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// A rejected start (missing idempotency key) is authenticated but never
// applied: no teardown.
glove::control::receipt_control_outcome rejected_outcome;
auto rejected_frame = (*protocol)->handle_frame(
make_request("start-4", "start_session", start_payload), 1'000, &rejected_outcome
);
REQUIRE(rejected_frame.has_value());
auto rejected_response = decode_response(*rejected_frame);
REQUIRE(rejected_response.has_value());
REQUIRE(rejected_response->error.has_value());
REQUIRE(!rejected_outcome.applied);
glove::control::detail::degrade_failed_delivery(config, rejected_outcome, "timeout");
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-9"});
// Audit journal: one event per degrade call, naming method and session
// where the typed decode succeeded; a replayed start's event reflects
// the replay disposition.
auto events = audit_sink->take();
REQUIRE(events.size() == 6);
REQUIRE(events[0].tool_name == "start_session:session-9");
REQUIRE(events[1].tool_name == "start_session:replay:session-9");
REQUIRE(events[2].tool_name == "start_session");
REQUIRE(events[3].tool_name == "start_session");
REQUIRE(events[4].tool_name == "health");
REQUIRE(events[5].tool_name == "start_session");
for (const auto& event : events) {
REQUIRE(event.what == glove::audit::action::control);
REQUIRE(event.status == glove::mcp::tool_call_status::transport_error);
}
// A failed stop must stay non-fatal and still be audited.
auto failing_sink = glove::audit::make_memory_sink();
auto failing_runtime = std::make_shared<degrade_test_runtime>(false);
const glove::control::receipt_audit_unix_server_config failing_config{
.socket_path = {},
.bootstrap_secret_path = {},
.producer = producer_config,
.plan_validator = validator,
.sessions = registry,
.runtime = failing_runtime,
.local_services = {},
.path_exposures = {},
.materialization_root = {},
.io_timeout_ms = 5'000,
.control_audit = failing_sink,
};
glove::control::detail::degrade_failed_delivery(
failing_config, outcome, "write control frame: Broken pipe"
);
REQUIRE(failing_runtime->stopped == std::vector<std::string>{"session-9"});
auto failing_events = failing_sink->take();
REQUIRE(failing_events.size() == 2);
REQUIRE(failing_events[0].tool_name == "start_session:session-9");
REQUIRE(failing_events[1].tool_name == "degrade_stop:session-9");
REQUIRE(failing_events[1].status == glove::mcp::tool_call_status::transport_error);
return 0;
}
// Regression test (Workflow #218): gloved exited fatally when the control
// client vanished between sending its request and reading the response. The
// response write must fail as a connection-scoped outcome — the daemon keeps
// serving — with the delivery failure recorded in the structured control
// audit journal. Read timeouts and listener failures remain errors.
//
// This drives the REAL server through create() and a genuinely
// authenticated+applied start_session through the real handle_frame, so the
// post-construction runtime ownership is exercised: on the pre-fix code the
// server's stored config held a moved-from runtime pointer and the guest was
// never torn down. Determinism: the client resets the socket only after the
// mutex-protected gate handshake proves the dispatch is parked inside the
// blocked gate, so the server is guaranteed to fail at the response write.
auto verify_broken_pipe_delivery_is_connection_scoped() -> int {
temporary_directory temp;
REQUIRE(!temp.root().empty());
const auto socket_path = temp.root() / "gloved.sock";
const auto secret_path = temp.root() / "bootstrap-secret";
const auto audit_key_path = temp.root() / "audit.key";
const auto journal_path = temp.root() / "receipts.journal";
const auto sessions_path = temp.root() / "sessions.journal";
const auto plan_source = temp.root() / "plan-source";
REQUIRE(std::filesystem::create_directory(plan_source));
std::ofstream{plan_source / "tracked.txt"} << "host-owned\n";
REQUIRE(write_owner_only(secret_path, bootstrap_secret));
REQUIRE(write_owner_only(audit_key_path, audit_key));
auto validator_result = plan_validator_for(plan_source);
REQUIRE(validator_result.has_value());
auto validator = std::make_shared<const glove::supervisor::session_plan_validator>(
std::move(*validator_result)
);
auto sessions = glove::control::session_registry::open_or_create(sessions_path, validator);
REQUIRE(sessions.has_value());
auto registry = std::shared_ptr<glove::control::session_registry>{std::move(*sessions)};
const glove::container::receipt_audit_producer_config producer_config{
.key_path = audit_key_path,
.journal_path = journal_path,
};
auto producer = glove::container::receipt_audit_producer::initialize(producer_config);
REQUIRE(producer.has_value());
REQUIRE((*producer)->acknowledge_bootstrap((*producer)->anchor()).has_value());
producer->reset();
auto audit_sink = glove::audit::make_memory_sink();
auto runtime = std::make_shared<degrade_test_runtime>();
const glove::control::receipt_audit_unix_server_config server_config{
.socket_path = socket_path,
.bootstrap_secret_path = secret_path,
.producer = producer_config,
.plan_validator = validator,
.sessions = registry,
.runtime = runtime,
.local_services = {},
.path_exposures = {},
.materialization_root = (temp.root() / "materializations").string(),
.io_timeout_ms = 5'000,
.control_audit = audit_sink,
};
auto server = glove::control::receipt_audit_unix_server::create(server_config);
REQUIRE(server.has_value());
std::optional<glove::control::receipt_audit_serve_outcome> outcome;
std::optional<std::string> server_error;
std::thread server_thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
server_error = served.error();
} else {
outcome = *served;
}
}};
// Prime the receipt producer through the real dispatch (receipt paging):
// start_session requires an initialized, reconciled producer.
const glove::container::receipt_audit_anchor genesis{
.key_id = std::string{audit_key_id},
.sequence = 0,
.head_hmac = std::string(64, '0'),
};
auto genesis_json = glz::write_json(genesis);
REQUIRE(genesis_json.has_value());
auto primed = transact(
socket_path,
make_request(
"page-prime",
"verify_audit_chain",
"{\"sage_anchor\":" + *genesis_json + ",\"limit\":10}"
)
);
server_thread.join();
REQUIRE(!server_error.has_value());
REQUIRE(outcome.has_value());
REQUIRE(*outcome == glove::control::receipt_audit_serve_outcome::served);
REQUIRE(primed.has_value());
auto primed_response = decode_response(*primed);
REQUIRE(primed_response.has_value());
REQUIRE(primed_response->result.has_value());
REQUIRE(audit_sink->take().empty());
outcome.reset();
server_thread = std::thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
server_error = served.error();
} else {
outcome = *served;
}
}};
const auto start_payload =
R"({"authorization":{"schema_version":1,"authorization_id":"auth-e2e",)"
R"("session_id":"session-degraded","controller_plan_digest":")" +
std::string(plan_digest) + R"(","plan_content_digest":")" + std::string(plan_digest) +
R"(","approved_at_ms":1,"expires_at_ms":4102444800000}})";
auto descriptor = connect_to(socket_path);
REQUIRE(descriptor.get() >= 0);
// Arm the start gate: the runtime's start() will block inside the real
// handle_frame dispatch until we release it after tearing the connection
// down, guaranteeing the server's response write hits a dead peer.
runtime->arm_start_gate();
const auto frame =
make_request("start-e2e", "start_session", start_payload, "degrade-start-e2e");
const auto size = htonl(static_cast<std::uint32_t>(frame.size()));
REQUIRE(write_exact(descriptor.get(), &size, sizeof(size)));
REQUIRE(write_exact(descriptor.get(), frame.data(), frame.size()));
// Synchronize on the gate handshake: the condition variable proves the
// dispatch is parked inside the blocked gate (inside handle_frame,
// before the start has been applied) BEFORE the reset, so the server's
// response write failure is deterministic rather than racy with the
// read. The started snapshot is only recorded after start() returns, so
// it is not usable as a "dispatch is inside the gate" observable.
REQUIRE(runtime->wait_for_start_gate(std::chrono::seconds{5}));
const ::linger reset_on_close{.l_onoff = 1, .l_linger = 0};
REQUIRE(
::setsockopt(
descriptor.get(), SOL_SOCKET, SO_LINGER, &reset_on_close, sizeof(reset_on_close)
) == 0
);
descriptor.reset();
// Now the peer is gone; let the applied start complete so the server
// fails at the response write, deterministically.
runtime->release_start_gate();
server_thread.join();
REQUIRE(!server_error.has_value());
REQUIRE(outcome.has_value());
REQUIRE(*outcome == glove::control::receipt_audit_serve_outcome::connection_failed);
// The HIGH blocker under repair: the degrade path must hold the SAME live
// runtime the protocol dispatched through, so the guest launched by the
// applied start_session is torn down despite the vanished controller.
REQUIRE(runtime->started_snapshot() == std::vector<std::string>{"session-degraded"});
REQUIRE(runtime->stopped_snapshot() == std::vector<std::string>{"session-degraded"});
const auto events = audit_sink->take();
REQUIRE(events.size() == 1);
REQUIRE(events.front().what == glove::audit::action::control);
REQUIRE(events.front().tool_name == "start_session:session-degraded");
REQUIRE(events.front().status == glove::mcp::tool_call_status::transport_error);
// The daemon-level invariant under repair: the server keeps serving
// subsequent connections after a connection-scoped delivery failure.
std::optional<std::string> recovery_error;
std::thread recovery_thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
recovery_error = served.error();
}
}};
auto recovery = transact(socket_path, make_request("health-2", "health", "null"));
recovery_thread.join();
REQUIRE(!recovery_error.has_value());
REQUIRE(recovery.has_value());
auto recovery_response = decode_response(*recovery);
REQUIRE(recovery_response.has_value());
REQUIRE(recovery_response->result.has_value());
return 0;
}
// Regression test (Workflow #218, round 3): an authenticated idempotent
// start_session replay is a success response but never a freshly applied
// launch. If the transport fails while a replay is being served, the live
// guest — launched by the earlier, already-delivered request — must NOT be
// stopped, outcome.applied must stay false, and the control audit event
// must reflect the replay disposition. The replay dispatch is parked inside
// a mutex-protected gate and observed via a condition-variable handshake
// before the client resets, so the broken pipe is deterministic.
auto verify_replay_broken_pipe_never_stops_live_session() -> int {
temporary_directory temp;
REQUIRE(!temp.root().empty());
const auto socket_path = temp.root() / "gloved.sock";
const auto secret_path = temp.root() / "bootstrap-secret";
const auto audit_key_path = temp.root() / "audit.key";
const auto journal_path = temp.root() / "receipts.journal";
const auto sessions_path = temp.root() / "sessions.journal";
const auto plan_source = temp.root() / "plan-source";
REQUIRE(std::filesystem::create_directory(plan_source));
std::ofstream{plan_source / "tracked.txt"} << "host-owned\n";
REQUIRE(write_owner_only(secret_path, bootstrap_secret));
REQUIRE(write_owner_only(audit_key_path, audit_key));
auto validator_result = plan_validator_for(plan_source);
REQUIRE(validator_result.has_value());
auto validator = std::make_shared<const glove::supervisor::session_plan_validator>(
std::move(*validator_result)
);
auto sessions = glove::control::session_registry::open_or_create(sessions_path, validator);
REQUIRE(sessions.has_value());
auto registry = std::shared_ptr<glove::control::session_registry>{std::move(*sessions)};
const glove::container::receipt_audit_producer_config producer_config{
.key_path = audit_key_path,
.journal_path = journal_path,
};
auto producer = glove::container::receipt_audit_producer::initialize(producer_config);
REQUIRE(producer.has_value());
REQUIRE((*producer)->acknowledge_bootstrap((*producer)->anchor()).has_value());
producer->reset();
auto audit_sink = glove::audit::make_memory_sink();
auto runtime = std::make_shared<degrade_test_runtime>();
const glove::control::receipt_audit_unix_server_config server_config{
.socket_path = socket_path,
.bootstrap_secret_path = secret_path,
.producer = producer_config,
.plan_validator = validator,
.sessions = registry,
.runtime = runtime,
.local_services = {},
.path_exposures = {},
.materialization_root = (temp.root() / "materializations").string(),
.io_timeout_ms = 5'000,
.control_audit = audit_sink,
};
auto server = glove::control::receipt_audit_unix_server::create(server_config);
REQUIRE(server.has_value());
const glove::container::receipt_audit_anchor genesis{
.key_id = std::string{audit_key_id},
.sequence = 0,
.head_hmac = std::string(64, '0'),
};
auto genesis_json = glz::write_json(genesis);
REQUIRE(genesis_json.has_value());
const auto start_payload =
R"({"authorization":{"schema_version":1,"authorization_id":"auth-replay",)"
R"("session_id":"session-replayed","controller_plan_digest":")" +
std::string(plan_digest) + R"(","plan_content_digest":")" + std::string(plan_digest) +
R"(","approved_at_ms":1,"expires_at_ms":4102444800000}})";
// Each delivered connection pairs one serve_one() with one transact().
auto serve_and_transact = [&](std::string_view frame) -> std::optional<std::string> {
std::optional<std::string> server_failure;
std::thread server_thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
server_failure = served.error();
}
}};
auto response = transact(socket_path, frame);
server_thread.join();
// REQUIRE's `return 1` does not fit this lambda's optional return
// type, so failures exit the test process directly.
if (server_failure.has_value() || !response.has_value()) {
std::fprintf(
stderr, "REQUIRE failed: serve_and_transact @ %s:%d\n", __FILE__, __LINE__
);
std::exit(1);
}
return response;
};
// Connection 1: prime the producer and deliver a genuine fresh start.
auto primed = serve_and_transact(make_request(
"page-prime", "verify_audit_chain", "{\"sage_anchor\":" + *genesis_json + ",\"limit\":10}"
));
auto primed_response = decode_response(*primed);
REQUIRE(primed_response.has_value());
REQUIRE(primed_response->result.has_value());
auto delivered = serve_and_transact(
make_request("start-live", "start_session", start_payload, "replay-start-e2e")
);
auto delivered_response = decode_response(*delivered);
REQUIRE(delivered_response.has_value());
REQUIRE(delivered_response->result.has_value());
REQUIRE(runtime->started_snapshot() == std::vector<std::string>{"session-replayed"});
REQUIRE(runtime->stopped_snapshot().empty());
REQUIRE(audit_sink->take().empty());
// Connection 2: authenticated idempotent replay of the same request.
std::optional<glove::control::receipt_audit_serve_outcome> outcome;
std::optional<std::string> server_error;
std::thread server_thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
server_error = served.error();
} else {
outcome = *served;
}
}};
auto descriptor = connect_to(socket_path);
REQUIRE(descriptor.get() >= 0);
runtime->arm_start_gate();
const auto frame =
make_request("start-replay", "start_session", start_payload, "replay-start-e2e");
const auto size = htonl(static_cast<std::uint32_t>(frame.size()));
REQUIRE(write_exact(descriptor.get(), &size, sizeof(size)));
REQUIRE(write_exact(descriptor.get(), frame.data(), frame.size()));
// The gate handshake proves the replay dispatch is parked inside the
// blocked gate before the connection is reset.
REQUIRE(runtime->wait_for_start_gate(std::chrono::seconds{5}));
const ::linger reset_on_close{.l_onoff = 1, .l_linger = 0};
REQUIRE(
::setsockopt(
descriptor.get(), SOL_SOCKET, SO_LINGER, &reset_on_close, sizeof(reset_on_close)
) == 0
);
descriptor.reset();
runtime->release_start_gate();
server_thread.join();
REQUIRE(!server_error.has_value());
REQUIRE(outcome.has_value());
REQUIRE(*outcome == glove::control::receipt_audit_serve_outcome::connection_failed);
// The replay must not be treated as a fresh launch: the live session is
// still running (no stop), and the replay's delivery failure produced
// exactly one control audit event carrying the replay disposition.
const std::vector<std::string> expected_started{"session-replayed", "session-replayed"};
REQUIRE(runtime->started_snapshot() == expected_started);
REQUIRE(runtime->stopped_snapshot().empty());
const auto events = audit_sink->take();
REQUIRE(events.size() == 1);
REQUIRE(events.front().what == glove::audit::action::control);
REQUIRE(events.front().tool_name == "start_session:replay:session-replayed");
REQUIRE(events.front().status == glove::mcp::tool_call_status::transport_error);
// The daemon keeps serving after the connection-scoped failure.
std::optional<std::string> recovery_error;
std::thread recovery_thread{[&] {
if (auto served = (*server)->serve_one(); !served) {
recovery_error = served.error();
}
}};
auto recovery = transact(socket_path, make_request("health-3", "health", "null"));
recovery_thread.join();
REQUIRE(!recovery_error.has_value());
REQUIRE(recovery.has_value());
auto recovery_response = decode_response(*recovery);
REQUIRE(recovery_response.has_value());
REQUIRE(recovery_response->result.has_value());
return 0;
}
auto verify_peer_credential_contract() -> int {
int descriptors[2] = {-1, -1};
REQUIRE(::socketpair(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0, descriptors) == 0);
const unique_fd local{descriptors[0]};
const unique_fd peer{descriptors[1]};
const auto owner = ::geteuid();
REQUIRE(glove::control::detail::verify_peer_owner(local.get(), owner).has_value());
const auto wrong_owner = owner == std::numeric_limits<::uid_t>::max() ? owner - 1U : owner + 1U;
REQUIRE(!glove::control::detail::verify_peer_owner(local.get(), wrong_owner).has_value());
return 0;
}
auto run() -> int {
REQUIRE(verify_peer_credential_contract() == 0);
REQUIRE(verify_degrade_requires_authenticated_applied_outcome() == 0);
REQUIRE(verify_broken_pipe_delivery_is_connection_scoped() == 0);
REQUIRE(verify_replay_broken_pipe_never_stops_live_session() == 0);
temporary_directory temp;
REQUIRE(!temp.root().empty());
const auto socket_path = temp.root() / "gloved.sock";
const auto secret_path = temp.root() / "bootstrap-secret";
const auto audit_key_path = temp.root() / "audit.key";
const auto journal_path = temp.root() / "receipts.journal";
REQUIRE(write_owner_only(secret_path, bootstrap_secret));
REQUIRE(write_owner_only(audit_key_path, audit_key));
const glove::container::receipt_audit_anchor genesis{
.key_id = std::string{audit_key_id},
.sequence = 0,
.head_hmac = std::string(64, '0'),
};
const glove::container::receipt_audit_producer_config producer_config{
.key_path = audit_key_path,
.journal_path = journal_path,
};
auto producer = glove::container::receipt_audit_producer::initialize(producer_config);
REQUIRE(producer.has_value());
REQUIRE((*producer)->anchor() == genesis);
REQUIRE((*producer)->acknowledge_bootstrap(genesis).has_value());
auto reservation = (*producer)->reserve_terminal();
REQUIRE(reservation.has_value());
auto terminal =
(*producer)->commit_terminal(std::move(*reservation), "session-1", plan_digest, receipt());
REQUIRE(terminal.has_value());
const auto terminal_anchor = (*producer)->anchor();
producer->reset();
const glove::control::receipt_audit_unix_server_config server_config{
.socket_path = socket_path,
.bootstrap_secret_path = secret_path,
.producer = producer_config,
.plan_validator = {},
.sessions = {},
.runtime = {},
.local_services = {},
.path_exposures = {},
.materialization_root = {},
.io_timeout_ms = 100,
.control_audit = {},
};
auto server = glove::control::receipt_audit_unix_server::create(server_config);
REQUIRE(server.has_value());
struct stat socket_status{};
REQUIRE(::lstat(socket_path.c_str(), &socket_status) == 0);
REQUIRE(S_ISSOCK(socket_status.st_mode));
REQUIRE(socket_status.st_uid == ::geteuid());
REQUIRE((socket_status.st_mode & 0777U) == 0600U);
std::optional<std::string> server_error;
std::thread server_thread{[&] {
for (std::size_t request = 0; request < 2; ++request) {
if (auto served = (*server)->serve_one(); !served) {
server_error = served.error();
return;
}
}
}};
const auto genesis_json = glz::write_json(genesis);
REQUIRE(genesis_json.has_value());
const auto page_payload = "{\"sage_anchor\":" + *genesis_json + ",\"limit\":1000}";
auto page_frame =
transact(socket_path, make_request("page-1", "verify_audit_chain", page_payload));
REQUIRE(page_frame.has_value());
auto page_response = decode_response(*page_frame);
REQUIRE(page_response.has_value());
REQUIRE(page_response->result.has_value());
wire_test::page_result page;
REQUIRE(!glz::read<glz::opts{.error_on_unknown_keys = true}>(page, page_response->result->str));
REQUIRE(page.envelopes.size() == 1);
REQUIRE(page.envelopes.front() == *terminal);
REQUIRE(page.local_anchor == terminal_anchor);
const auto terminal_anchor_json = glz::write_json(terminal_anchor);
REQUIRE(terminal_anchor_json.has_value());
const auto ack_payload = "{\"anchor\":" + *terminal_anchor_json + "}";
auto ack_frame = transact(
socket_path,
make_request("ack-1", "acknowledge_audit_chain", ack_payload, "receipt-audit-ack-1")
);
REQUIRE(ack_frame.has_value());
auto ack_response = decode_response(*ack_frame);
REQUIRE(ack_response.has_value());
REQUIRE(ack_response->result.has_value());
server_thread.join();
REQUIRE(!server_error.has_value());
server->reset();
REQUIRE(::access(socket_path.c_str(), F_OK) != 0);
REQUIRE(::chmod(secret_path.c_str(), 0644) == 0);
REQUIRE(!glove::control::receipt_audit_unix_server::create(server_config).has_value());
REQUIRE(::chmod(secret_path.c_str(), 0600) == 0);
REQUIRE(::chmod(temp.root().c_str(), 0755) == 0);
REQUIRE(!glove::control::receipt_audit_unix_server::create(server_config).has_value());
REQUIRE(::chmod(temp.root().c_str(), 0700) == 0);
auto timeout_server = glove::control::receipt_audit_unix_server::create(server_config);
REQUIRE(timeout_server.has_value());
std::optional<std::string> timeout_error;
std::thread timeout_thread{[&] {
auto served = (*timeout_server)->serve_one();
if (!served) {
timeout_error = served.error();
}
}};
auto stalled = connect_to(socket_path);
REQUIRE(stalled.get() >= 0);
timeout_thread.join();
REQUIRE(timeout_error.has_value());
REQUIRE(timeout_error->find("timed out") != std::string::npos);
return 0;
}
} // namespace
auto main() -> int {
return run();
}
Reference in a new issue View git blame Copy permalink