mirror of
https://github.com/brazilofmux/tinymux
synced 2026-08-13 00:23:11 -04:00
52f8c08eb9
#1265: Bound WorkQueue to MAX_PENDING=1024 (blocking enqueue) and wait on futures in GameSession reader so a stream cannot flood the main loop. #1266: Cap concurrent subscribers per session (MAX_SUBSCRIBERS=8); reject GameSession/Subscribe/SubscribeGmcp/WS with RESOURCE_EXHAUSTED / close. #1267: grpc-web SendInput routes through TelnetBridge::convertInput like native gRPC and WS GameSession. #1268: Reject gRPC/WS/grpc-web input lines above MAX_INPUT_LINE_LENGTH (8192, shared with front-door telnet assembly); drop oversized GMCP. #1269: GetGameStatus returns host PIDs only for admin accounts; any authenticated session still sees running/up. #1270: proxy_regression covers subscriber cap, line-limit constants, work-queue cap constant, and convertInput non-UTF8 target.
1064 lines
45 KiB
C++
1064 lines
45 KiB
C++
#ifdef GRPC_ENABLED
|
|
|
|
#include "grpc_server.h"
|
|
#include "work_queue.h"
|
|
#include "session_manager.h"
|
|
#include "account_manager.h"
|
|
#include "process_manager.h"
|
|
#include "config.h"
|
|
#include "hydra_log.h"
|
|
#include "telnet_utils.h"
|
|
#include "utf8_utils.h"
|
|
|
|
#include "hydra.grpc.pb.h"
|
|
|
|
#include <grpcpp/grpcpp.h>
|
|
#include <grpcpp/server_builder.h>
|
|
#include <chrono>
|
|
#include <fstream>
|
|
|
|
using grpc::Server;
|
|
using grpc::ServerBuilder;
|
|
using grpc::ServerContext;
|
|
using grpc::ServerWriter;
|
|
using grpc::ServerReaderWriter;
|
|
using grpc::Status;
|
|
using grpc::StatusCode;
|
|
|
|
// ---- Auth helper: extract session_id from metadata or message field ----
|
|
|
|
static std::string getSessionId(ServerContext* ctx, const std::string& msgField) {
|
|
// Prefer metadata "authorization" header
|
|
auto md = ctx->client_metadata();
|
|
auto it = md.find("authorization");
|
|
if (it != md.end()) {
|
|
return std::string(it->second.data(), it->second.size());
|
|
}
|
|
// Fall back to per-message session_id field
|
|
return msgField;
|
|
}
|
|
|
|
// #1268: match front-door telnet line assembly.
|
|
static bool inputLineWithinLimit(const std::string& line) {
|
|
return line.size() <= HydraSession::MAX_INPUT_LINE_LENGTH;
|
|
}
|
|
|
|
// ---- Helper: map C++ LinkState to proto LinkState enum ----
|
|
|
|
static hydra::LinkState toProtoLinkState(LinkState s) {
|
|
switch (s) {
|
|
case LinkState::Connecting: return hydra::LINK_CONNECTING;
|
|
case LinkState::TlsHandshaking: return hydra::LINK_TLS;
|
|
case LinkState::Negotiating: return hydra::LINK_NEGOTIATING;
|
|
case LinkState::AutoLoggingIn: return hydra::LINK_LOGGING_IN;
|
|
case LinkState::Active: return hydra::LINK_ACTIVE;
|
|
case LinkState::Reconnecting: return hydra::LINK_RECONNECTING;
|
|
case LinkState::Suspended: return hydra::LINK_SUSPENDED;
|
|
case LinkState::Dead: return hydra::LINK_DEAD;
|
|
}
|
|
return hydra::LINK_UNKNOWN;
|
|
}
|
|
|
|
static hydra::SessionState toProtoSessionState(SessionState s) {
|
|
switch (s) {
|
|
case SessionState::Login: return hydra::SESSION_LOGIN;
|
|
case SessionState::Active: return hydra::SESSION_ACTIVE;
|
|
case SessionState::Detached: return hydra::SESSION_DETACHED;
|
|
}
|
|
return hydra::SESSION_UNKNOWN;
|
|
}
|
|
|
|
// ---- Helper: populate LinkInfo proto ----
|
|
|
|
static void fillLinkInfo(hydra::LinkInfo* li, const BackDoorLink& link,
|
|
size_t idx, size_t activeIdx) {
|
|
li->set_number(static_cast<int>(idx) + 1);
|
|
li->set_game_name(link.gameName);
|
|
li->set_character(link.character);
|
|
li->set_active(idx == activeIdx);
|
|
li->set_gmcp_enabled(link.gmcpEnabled);
|
|
li->set_retry_count(link.retryCount);
|
|
li->set_next_retry(static_cast<int64_t>(link.nextRetry));
|
|
li->set_state(toProtoLinkState(link.state));
|
|
}
|
|
|
|
class HydraServiceImpl final : public hydra::HydraService::Service {
|
|
public:
|
|
HydraServiceImpl(SessionManager& sessionMgr, AccountManager& accounts,
|
|
const HydraConfig& config, ProcessManager& procMgr,
|
|
WorkQueue& workQueue)
|
|
: sessionMgr_(sessionMgr), accounts_(accounts), config_(config),
|
|
procMgr_(procMgr), workQueue_(workQueue) {}
|
|
|
|
// ---- Authentication ----
|
|
|
|
Status Authenticate(ServerContext* ctx, const hydra::AuthRequest* req,
|
|
hydra::AuthResponse* resp) override {
|
|
// #1097: extract peer IP so gRPC cannot bypass front-door lockout.
|
|
std::string peer = ctx->peer();
|
|
std::string clientIp;
|
|
{
|
|
auto c1 = peer.find(':');
|
|
if (c1 != std::string::npos) {
|
|
auto c2 = peer.rfind(':');
|
|
if (c2 > c1) clientIp = peer.substr(c1 + 1, c2 - c1 - 1);
|
|
else clientIp = peer.substr(c1 + 1);
|
|
}
|
|
}
|
|
auto future = workQueue_.enqueue<std::string>(
|
|
[user = req->username(), pw = req->password(), clientIp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
return sm.authenticateAndGetSession(user, pw, clientIp);
|
|
});
|
|
std::string pid = future.get();
|
|
if (pid.empty()) {
|
|
resp->set_success(false);
|
|
resp->set_error("authentication failed");
|
|
} else {
|
|
resp->set_success(true);
|
|
resp->set_session_id(pid);
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
Status CreateAccount(ServerContext* ctx, const hydra::CreateAccountRequest* req,
|
|
hydra::CreateAccountResponse* resp) override {
|
|
// Extract client IP from gRPC peer (format: "ipv4:1.2.3.4:port")
|
|
std::string peer = ctx->peer();
|
|
std::string clientIp;
|
|
{
|
|
auto c1 = peer.find(':');
|
|
if (c1 != std::string::npos) {
|
|
auto c2 = peer.rfind(':');
|
|
if (c2 > c1) clientIp = peer.substr(c1 + 1, c2 - c1 - 1);
|
|
else clientIp = peer.substr(c1 + 1);
|
|
}
|
|
}
|
|
auto future = workQueue_.enqueue<std::pair<std::string, std::string>>(
|
|
[user = req->username(), pw = req->password(), clientIp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&)
|
|
-> std::pair<std::string, std::string> {
|
|
std::string errorMsg;
|
|
std::string pid = sm.createAccountAndGetSession(user, pw, clientIp, errorMsg);
|
|
return {pid, errorMsg};
|
|
});
|
|
auto [pid, err] = future.get();
|
|
if (pid.empty()) {
|
|
resp->set_success(false);
|
|
resp->set_error(err.empty() ? "account creation failed" : err);
|
|
} else {
|
|
resp->set_success(true);
|
|
resp->set_session_id(pid);
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Session lifecycle ----
|
|
|
|
Status GetSession(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
hydra::SessionInfo* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
resp->set_session_id(s->persistId);
|
|
resp->set_username(s->username);
|
|
resp->set_active_link(s->links.empty() ? 0
|
|
: static_cast<int>(s->activeLink) + 1);
|
|
resp->set_scrollback_lines(static_cast<int>(s->scrollback.count()));
|
|
resp->set_state(toProtoSessionState(s->state));
|
|
resp->set_created(static_cast<int64_t>(s->created));
|
|
resp->set_last_activity(static_cast<int64_t>(s->lastActivity));
|
|
for (size_t i = 0; i < s->links.size(); i++) {
|
|
fillLinkInfo(resp->add_links(), s->links[i], i, s->activeLink);
|
|
}
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
Status DetachSession(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
hydra::Empty* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
s->state = SessionState::Detached;
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
Status DestroySession(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
hydra::Empty* resp) override {
|
|
return Status(StatusCode::UNIMPLEMENTED, "use /quit via telnet for now");
|
|
}
|
|
|
|
Status Ping(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
hydra::Empty* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
s->lastActivity = time(nullptr);
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Game catalog ----
|
|
|
|
Status ListGames(ServerContext* ctx, const hydra::Empty* req,
|
|
hydra::GameList* resp) override {
|
|
// #1102: require a valid session token.
|
|
(void)req;
|
|
std::string sid = getSessionId(ctx, "");
|
|
auto authFuture = workQueue_.enqueue<bool>(
|
|
[sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
return sm.findByPersistId(sid) != nullptr;
|
|
});
|
|
if (sid.empty() || !authFuture.get()) {
|
|
return Status(StatusCode::UNAUTHENTICATED, "session required");
|
|
}
|
|
for (const auto& game : config_.games) {
|
|
auto* gi = resp->add_games();
|
|
gi->set_name(game.name);
|
|
gi->set_host(game.host);
|
|
gi->set_port(game.port);
|
|
gi->set_type(game.type == GameType::Local ? hydra::GAME_LOCAL : hydra::GAME_REMOTE);
|
|
gi->set_autostart(game.autostart);
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Link management ----
|
|
|
|
Status Connect(ServerContext* ctx, const hydra::ConnectRequest* req,
|
|
hydra::ConnectResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, game = req->game_name(), resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
size_t before = s->links.size();
|
|
sm.connectToGame(*s, game);
|
|
if (s->links.size() > before) {
|
|
resp->set_success(true);
|
|
resp->set_link_number(static_cast<int>(s->links.size()));
|
|
return true;
|
|
}
|
|
resp->set_error("connect failed");
|
|
return false;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status SwitchLink(ServerContext* ctx, const hydra::SwitchRequest* req,
|
|
hydra::SwitchResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, num = req->link_number(), resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
size_t idx = static_cast<size_t>(num - 1);
|
|
if (idx >= s->links.size()) {
|
|
resp->set_error("invalid link number");
|
|
return false;
|
|
}
|
|
s->activeLink = idx;
|
|
resp->set_success(true);
|
|
return true;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status DisconnectLink(ServerContext* ctx, const hydra::DisconnectRequest* req,
|
|
hydra::DisconnectResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, num = req->link_number(), resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
size_t idx = static_cast<size_t>(num - 1);
|
|
if (idx >= s->links.size()) {
|
|
resp->set_error("invalid link number");
|
|
return false;
|
|
}
|
|
sm.closeLink(*s, idx);
|
|
resp->set_success(true);
|
|
return true;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status ListLinks(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
hydra::LinkList* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
for (size_t i = 0; i < s->links.size(); i++) {
|
|
fillLinkInfo(resp->add_links(), s->links[i], i, s->activeLink);
|
|
}
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Bidirectional I/O (primary channel) ----
|
|
|
|
Status GameSession(ServerContext* ctx,
|
|
ServerReaderWriter<hydra::ServerMessage, hydra::ClientMessage>* stream) override {
|
|
// Read initial metadata for session_id and color_format
|
|
auto md = ctx->client_metadata();
|
|
std::string sid;
|
|
auto it = md.find("authorization");
|
|
if (it != md.end()) {
|
|
sid = std::string(it->second.data(), it->second.size());
|
|
}
|
|
// Also check "session-id" metadata as fallback
|
|
if (sid.empty()) {
|
|
it = md.find("session-id");
|
|
if (it != md.end()) {
|
|
sid = std::string(it->second.data(), it->second.size());
|
|
}
|
|
}
|
|
if (sid.empty()) {
|
|
return Status(StatusCode::UNAUTHENTICATED, "session_id required in metadata");
|
|
}
|
|
|
|
// Get output queue
|
|
std::shared_ptr<HydraSession::OutputQueue> oq;
|
|
auto future = workQueue_.enqueue<std::shared_ptr<HydraSession::OutputQueue>>(
|
|
[&sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&)
|
|
-> std::shared_ptr<HydraSession::OutputQueue> {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return nullptr;
|
|
s->state = SessionState::Active;
|
|
s->lastActivity = time(nullptr);
|
|
return s->outputQueue;
|
|
});
|
|
oq = future.get();
|
|
if (!oq) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
|
|
// Register as subscriber (wants both output and GMCP)
|
|
int subId;
|
|
std::shared_ptr<HydraSession::SubscriberQueue> sq;
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
auto [id, q] = oq->addSubscriber(true, true);
|
|
subId = id;
|
|
sq = q;
|
|
}
|
|
if (!sq) {
|
|
// #1266: too many concurrent GameSession/Subscribe streams.
|
|
return Status(StatusCode::RESOURCE_EXHAUSTED,
|
|
"too many subscribers on this session");
|
|
}
|
|
|
|
// Replay cached GMCP state into the new subscriber's queue
|
|
{
|
|
auto replayFuture = workQueue_.enqueue<bool>(
|
|
[sid, sq](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) -> bool {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
sm.replayGmcpCache(*s, sq);
|
|
return true;
|
|
});
|
|
replayFuture.get();
|
|
}
|
|
|
|
// Reader thread: process client input.
|
|
// #1265: wait on each future so this stream cannot flood the work
|
|
// queue faster than the main loop drains (bounded enqueue is the
|
|
// global backstop; future.get is per-stream sequencing).
|
|
std::atomic<bool> done{false};
|
|
std::thread reader([&]() {
|
|
hydra::ClientMessage msg;
|
|
while (stream->Read(&msg)) {
|
|
if (msg.has_input_line()) {
|
|
if (!inputLineWithinLimit(msg.input_line())) {
|
|
LOG_WARN("GameSession: input_line too long (%zu), dropped",
|
|
msg.input_line().size());
|
|
continue;
|
|
}
|
|
auto fut = workQueue_.enqueue<void>(
|
|
[sid, line = msg.input_line()]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return;
|
|
BackDoorLink* active = s->getActiveLink();
|
|
if (active && active->state == LinkState::Active) {
|
|
std::string data = TelnetBridge().convertInput(
|
|
ganl::EncodingType::Utf8,
|
|
active->protoState.encoding,
|
|
line);
|
|
data += "\r\n";
|
|
sm.safeWrite(active->handle, data);
|
|
}
|
|
});
|
|
fut.get();
|
|
} else if (msg.has_ping()) {
|
|
// Queue pong for the writer loop (don't write from reader thread)
|
|
HydraSession::OutputItem pongItem;
|
|
pongItem.puaText = "\x01PONG\x01" +
|
|
std::to_string(msg.ping().client_timestamp());
|
|
pongItem.source = "__pong__";
|
|
pongItem.timestamp = 0;
|
|
pongItem.linkNumber = 0;
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
sq->output.push(std::move(pongItem));
|
|
}
|
|
oq->cv.notify_all();
|
|
} else if (msg.has_preferences()) {
|
|
const auto& prefs = msg.preferences();
|
|
// Update subscriber's color format (under lock — writer reads it)
|
|
// Skip if COLOR_UNSPECIFIED (0) — means "don't change"
|
|
if (prefs.color_format() != hydra::COLOR_UNSPECIFIED) {
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
sq->renderFormat = static_cast<HydraSession::RenderFormat>(
|
|
prefs.color_format());
|
|
}
|
|
// Forward terminal size to game via NAWS
|
|
if (prefs.terminal_width() > 0 || prefs.terminal_height() > 0) {
|
|
auto fut = workQueue_.enqueue<void>(
|
|
[sid, w = prefs.terminal_width(), h = prefs.terminal_height()]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return;
|
|
BackDoorLink* active = s->getActiveLink();
|
|
if (active && active->handle != ganl::InvalidConnectionHandle) {
|
|
// Send NAWS to game: IAC SB NAWS w_hi w_lo h_hi h_lo IAC SE
|
|
uint16_t width = w ? static_cast<uint16_t>(w) : 80;
|
|
uint16_t height = h ? static_cast<uint16_t>(h) : 24;
|
|
sm.safeWrite(active->handle, buildNawsFrame(width, height));
|
|
}
|
|
});
|
|
fut.get();
|
|
}
|
|
// Store terminal_type in session for future TTYPE forwarding
|
|
if (!prefs.terminal_type().empty()) {
|
|
auto fut = workQueue_.enqueue<void>(
|
|
[sid, ttype = prefs.terminal_type()]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (s) s->terminalType = ttype;
|
|
});
|
|
fut.get();
|
|
}
|
|
LOG_DEBUG("GameSession: client set preferences color=%d width=%u height=%u type=%s",
|
|
prefs.color_format(), prefs.terminal_width(), prefs.terminal_height(),
|
|
prefs.terminal_type().c_str());
|
|
} else if (msg.has_gmcp()) {
|
|
// Cap GMCP package+json similarly to input lines (#1268).
|
|
if (msg.gmcp().package().size() + msg.gmcp().json().size()
|
|
> HydraSession::MAX_INPUT_LINE_LENGTH) {
|
|
LOG_WARN("GameSession: GMCP payload too long, dropped");
|
|
continue;
|
|
}
|
|
// Forward GMCP to active link
|
|
auto fut = workQueue_.enqueue<void>(
|
|
[sid, pkg = msg.gmcp().package(), json = msg.gmcp().json()]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return;
|
|
BackDoorLink* active = s->getActiveLink();
|
|
if (active && active->handle != ganl::InvalidConnectionHandle
|
|
&& active->gmcpEnabled) {
|
|
std::string payload = pkg + " " + json;
|
|
sm.safeWrite(active->handle, buildGmcpFrame(payload));
|
|
}
|
|
});
|
|
fut.get();
|
|
}
|
|
}
|
|
done.store(true);
|
|
oq->cv.notify_all();
|
|
});
|
|
|
|
// Writer loop: drain this subscriber's own queues
|
|
while (!done.load() && !ctx->IsCancelled()) {
|
|
std::unique_lock<std::mutex> lock(oq->mutex);
|
|
oq->cv.wait_for(lock, std::chrono::milliseconds(500),
|
|
[&sq, &done] { return !sq->output.empty() || !sq->gmcp.empty() || done.load(); });
|
|
|
|
// Read renderFormat under lock (writer reads, reader sets)
|
|
auto currentFmt = sq->renderFormat;
|
|
|
|
while (!sq->output.empty()) {
|
|
auto item = std::move(sq->output.front());
|
|
sq->output.pop();
|
|
lock.unlock();
|
|
|
|
// Check for pong sentinel (queued by reader thread)
|
|
if (item.source == "__pong__") {
|
|
hydra::ServerMessage msg;
|
|
auto* pong = msg.mutable_pong();
|
|
int64_t clientTs = 0;
|
|
try { clientTs = std::stoll(item.puaText.substr(6)); }
|
|
catch (...) {}
|
|
pong->set_client_timestamp(clientTs);
|
|
pong->set_server_timestamp(
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::system_clock::now().time_since_epoch()).count());
|
|
if (!stream->Write(msg)) { done.store(true); break; }
|
|
} else {
|
|
hydra::ServerMessage msg;
|
|
auto* go = msg.mutable_game_output();
|
|
go->set_text(sanitizeProtoTextForLog(
|
|
item.render(currentFmt), "grpc bidi output",
|
|
item.source, item.linkNumber));
|
|
go->set_source(item.source);
|
|
go->set_timestamp(static_cast<int64_t>(item.timestamp));
|
|
go->set_link_number(item.linkNumber);
|
|
go->set_end_of_record(item.endOfRecord);
|
|
if (!stream->Write(msg)) { done.store(true); break; }
|
|
}
|
|
lock.lock();
|
|
}
|
|
|
|
while (!sq->gmcp.empty()) {
|
|
auto item = std::move(sq->gmcp.front());
|
|
sq->gmcp.pop();
|
|
lock.unlock();
|
|
hydra::ServerMessage msg;
|
|
auto* gm = msg.mutable_gmcp();
|
|
gm->set_package(item.package);
|
|
gm->set_json(item.json);
|
|
gm->set_link_number(item.linkNumber);
|
|
if (!stream->Write(msg)) { done.store(true); break; }
|
|
lock.lock();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
oq->removeSubscriber(subId);
|
|
}
|
|
reader.join();
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Legacy unary I/O ----
|
|
|
|
Status SendInput(ServerContext* ctx, const hydra::InputRequest* req,
|
|
hydra::InputResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
// #1268: reject before enqueue so the work queue never holds a huge line.
|
|
if (!inputLineWithinLimit(req->line())) {
|
|
resp->set_success(false);
|
|
resp->set_error("line too long");
|
|
return Status::OK;
|
|
}
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, line = req->line(), resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
BackDoorLink* active = s->getActiveLink();
|
|
if (!active || active->state != LinkState::Active) {
|
|
resp->set_error("no active link");
|
|
return false;
|
|
}
|
|
std::string data = TelnetBridge().convertInput(
|
|
ganl::EncodingType::Utf8,
|
|
active->protoState.encoding,
|
|
line);
|
|
data += "\r\n";
|
|
sm.safeWrite(active->handle, data);
|
|
resp->set_success(true);
|
|
return true;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status Subscribe(ServerContext* ctx, const hydra::SessionRequest* req,
|
|
ServerWriter<hydra::GameOutput>* writer) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
std::shared_ptr<HydraSession::OutputQueue> oq;
|
|
auto future = workQueue_.enqueue<std::shared_ptr<HydraSession::OutputQueue>>(
|
|
[&sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&)
|
|
-> std::shared_ptr<HydraSession::OutputQueue> {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return nullptr;
|
|
return s->outputQueue;
|
|
});
|
|
oq = future.get();
|
|
if (!oq) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
|
|
// Forward terminal size to the active game link if provided
|
|
if (req->terminal_width() > 0 || req->terminal_height() > 0) {
|
|
uint32_t w = req->terminal_width();
|
|
uint32_t h = req->terminal_height();
|
|
workQueue_.enqueue<void>(
|
|
[sid, w, h]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return;
|
|
BackDoorLink* active = s->getActiveLink();
|
|
if (active && active->handle != ganl::InvalidConnectionHandle) {
|
|
uint16_t width = w ? static_cast<uint16_t>(w) : 80;
|
|
uint16_t height = h ? static_cast<uint16_t>(h) : 24;
|
|
sm.safeWrite(active->handle, buildNawsFrame(width, height));
|
|
}
|
|
});
|
|
}
|
|
|
|
// Register as subscriber (output only, no GMCP) with requested color format
|
|
// COLOR_UNSPECIFIED (0) → default to TrueColor
|
|
auto renderFmt = req->color_format() == hydra::COLOR_UNSPECIFIED
|
|
? HydraSession::RenderFormat::TrueColor
|
|
: static_cast<HydraSession::RenderFormat>(req->color_format());
|
|
int subId;
|
|
std::shared_ptr<HydraSession::SubscriberQueue> sq;
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
auto [id, q] = oq->addSubscriber(true, false, renderFmt);
|
|
subId = id;
|
|
sq = q;
|
|
}
|
|
if (!sq) {
|
|
return Status(StatusCode::RESOURCE_EXHAUSTED,
|
|
"too many subscribers on this session");
|
|
}
|
|
|
|
while (!ctx->IsCancelled()) {
|
|
std::unique_lock<std::mutex> lock(oq->mutex);
|
|
oq->cv.wait_for(lock, std::chrono::milliseconds(500),
|
|
[&sq] { return !sq->output.empty(); });
|
|
|
|
while (!sq->output.empty()) {
|
|
auto item = std::move(sq->output.front());
|
|
sq->output.pop();
|
|
lock.unlock();
|
|
hydra::GameOutput msg;
|
|
msg.set_text(sanitizeProtoTextForLog(
|
|
item.render(sq->renderFormat), "grpc subscribe output",
|
|
item.source, item.linkNumber));
|
|
msg.set_source(item.source);
|
|
msg.set_timestamp(static_cast<int64_t>(item.timestamp));
|
|
msg.set_link_number(item.linkNumber);
|
|
msg.set_end_of_record(item.endOfRecord);
|
|
if (!writer->Write(msg)) {
|
|
std::lock_guard<std::mutex> lk(oq->mutex);
|
|
oq->removeSubscriber(subId);
|
|
return Status::OK;
|
|
}
|
|
lock.lock();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
oq->removeSubscriber(subId);
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- GMCP subscription ----
|
|
|
|
Status SubscribeGmcp(ServerContext* ctx, const hydra::GmcpSubscribeRequest* req,
|
|
ServerWriter<hydra::GmcpMessage>* writer) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
std::shared_ptr<HydraSession::OutputQueue> oq;
|
|
auto future = workQueue_.enqueue<std::shared_ptr<HydraSession::OutputQueue>>(
|
|
[&sid](SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&)
|
|
-> std::shared_ptr<HydraSession::OutputQueue> {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return nullptr;
|
|
return s->outputQueue;
|
|
});
|
|
oq = future.get();
|
|
if (!oq) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
|
|
// Collect package filters
|
|
std::vector<std::string> filters;
|
|
for (int i = 0; i < req->packages_size(); i++) {
|
|
filters.push_back(req->packages(i));
|
|
}
|
|
|
|
// Register as subscriber (GMCP only, no output)
|
|
int subId;
|
|
std::shared_ptr<HydraSession::SubscriberQueue> sq;
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
auto [id, q] = oq->addSubscriber(false, true);
|
|
subId = id;
|
|
sq = q;
|
|
}
|
|
if (!sq) {
|
|
return Status(StatusCode::RESOURCE_EXHAUSTED,
|
|
"too many subscribers on this session");
|
|
}
|
|
|
|
while (!ctx->IsCancelled()) {
|
|
std::unique_lock<std::mutex> lock(oq->mutex);
|
|
oq->cv.wait_for(lock, std::chrono::milliseconds(500),
|
|
[&sq] { return !sq->gmcp.empty(); });
|
|
|
|
while (!sq->gmcp.empty()) {
|
|
auto item = std::move(sq->gmcp.front());
|
|
sq->gmcp.pop();
|
|
lock.unlock();
|
|
|
|
// Apply package filter if any
|
|
bool match = filters.empty();
|
|
if (!match) {
|
|
for (const auto& f : filters) {
|
|
if (item.package.compare(0, f.size(), f) == 0) {
|
|
match = true;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (match) {
|
|
hydra::GmcpMessage msg;
|
|
msg.set_package(item.package);
|
|
msg.set_json(item.json);
|
|
msg.set_link_number(item.linkNumber);
|
|
if (!writer->Write(msg)) {
|
|
std::lock_guard<std::mutex> lk(oq->mutex);
|
|
oq->removeSubscriber(subId);
|
|
return Status::OK;
|
|
}
|
|
}
|
|
lock.lock();
|
|
}
|
|
}
|
|
|
|
{
|
|
std::lock_guard<std::mutex> lock(oq->mutex);
|
|
oq->removeSubscriber(subId);
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Scroll-back ----
|
|
|
|
Status GetScrollBack(ServerContext* ctx, const hydra::ScrollBackRequest* req,
|
|
hydra::ScrollBackResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
// Resolve color format: UNSPECIFIED → TrueColor
|
|
auto fmt = req->color_format() == hydra::COLOR_UNSPECIFIED
|
|
? HydraSession::RenderFormat::TrueColor
|
|
: static_cast<HydraSession::RenderFormat>(req->color_format());
|
|
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, maxLines = req->max_lines(), fmt, resp]
|
|
(SessionManager& sm, AccountManager&, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
size_t n = maxLines > 0 ? static_cast<size_t>(maxLines)
|
|
: s->scrollback.count();
|
|
struct ReplayCtx {
|
|
hydra::ScrollBackResponse* resp;
|
|
HydraSession::RenderFormat fmt;
|
|
};
|
|
ReplayCtx rctx{resp, fmt};
|
|
s->scrollback.replay(n,
|
|
[](const std::string& text, const std::string& source,
|
|
time_t timestamp, void* ctx) {
|
|
auto* rc = static_cast<ReplayCtx*>(ctx);
|
|
auto* line = rc->resp->add_lines();
|
|
// Render PUA text at the requested color format
|
|
HydraSession::OutputItem item;
|
|
item.puaText = text;
|
|
line->set_text(sanitizeProtoTextForLog(
|
|
item.render(rc->fmt), "grpc scrollback",
|
|
source, 0));
|
|
line->set_source(source);
|
|
line->set_timestamp(static_cast<int64_t>(timestamp));
|
|
},
|
|
&rctx);
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Credentials ----
|
|
|
|
Status AddCredential(ServerContext* ctx, const hydra::AddCredentialRequest* req,
|
|
hydra::AddCredentialResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, game = req->game(), character = req->character(),
|
|
verb = req->verb(), name = req->name(), secret = req->secret(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
std::string errorMsg;
|
|
if (am.storeCredential(s->accountId, game, character,
|
|
verb, name, secret, errorMsg)) {
|
|
resp->set_success(true);
|
|
return true;
|
|
}
|
|
resp->set_error(errorMsg);
|
|
return false;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status DeleteCredential(ServerContext* ctx, const hydra::DeleteCredentialRequest* req,
|
|
hydra::DeleteCredentialResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, game = req->game(), character = req->character(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) { resp->set_error("session not found"); return false; }
|
|
resp->set_success(am.deleteCredential(s->accountId, game, character));
|
|
return true;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status ListCredentials(ServerContext* ctx, const hydra::ListCredentialsRequest* req,
|
|
hydra::ListCredentialsResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, req->session_id());
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig&, ProcessManager&) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s) return false;
|
|
auto creds = am.listCredentials(s->accountId);
|
|
for (const auto& c : creds) {
|
|
auto* sc = resp->add_credentials();
|
|
sc->set_game(c.game);
|
|
sc->set_character(c.character);
|
|
sc->set_verb(c.verb);
|
|
sc->set_name(c.name);
|
|
sc->set_auto_login(c.autoLogin);
|
|
}
|
|
return true;
|
|
});
|
|
if (!future.get()) return Status(StatusCode::NOT_FOUND, "session not found");
|
|
return Status::OK;
|
|
}
|
|
|
|
// ---- Process management ----
|
|
|
|
Status StartGame(ServerContext* ctx, const hydra::GameRequest* req,
|
|
hydra::GameResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, "");
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, gameName = req->game_name(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig& cfg, ProcessManager& pm) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s || !am.isAdmin(s->accountId)) {
|
|
resp->set_error("permission denied (admin required)");
|
|
return false;
|
|
}
|
|
const GameConfig* game = nullptr;
|
|
for (const auto& g : cfg.games) {
|
|
if (g.name == gameName) { game = &g; break; }
|
|
}
|
|
if (!game) { resp->set_error("unknown game"); return false; }
|
|
if (game->type != GameType::Local) {
|
|
resp->set_error("not a local game");
|
|
return false;
|
|
}
|
|
std::string errorMsg;
|
|
if (pm.startGame(*game, errorMsg)) {
|
|
resp->set_success(true);
|
|
resp->set_pid(static_cast<int>(pm.getPid(gameName)));
|
|
return true;
|
|
}
|
|
resp->set_error(errorMsg);
|
|
return false;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status StopGame(ServerContext* ctx, const hydra::GameRequest* req,
|
|
hydra::GameResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, "");
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, gameName = req->game_name(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig&, ProcessManager& pm) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s || !am.isAdmin(s->accountId)) {
|
|
resp->set_error("permission denied (admin required)");
|
|
return false;
|
|
}
|
|
resp->set_success(pm.stopGame(gameName));
|
|
if (!resp->success()) resp->set_error("game not running");
|
|
return true;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status RestartGame(ServerContext* ctx, const hydra::GameRequest* req,
|
|
hydra::GameResponse* resp) override {
|
|
std::string sid = getSessionId(ctx, "");
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, gameName = req->game_name(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig& cfg, ProcessManager& pm) {
|
|
HydraSession* s = sm.findByPersistId(sid);
|
|
if (!s || !am.isAdmin(s->accountId)) {
|
|
resp->set_error("permission denied (admin required)");
|
|
return false;
|
|
}
|
|
const GameConfig* game = nullptr;
|
|
for (const auto& g : cfg.games) {
|
|
if (g.name == gameName) { game = &g; break; }
|
|
}
|
|
if (!game) { resp->set_error("unknown game"); return false; }
|
|
std::string errorMsg;
|
|
if (pm.restartGame(*game, errorMsg)) {
|
|
resp->set_success(true);
|
|
resp->set_pid(static_cast<int>(pm.getPid(gameName)));
|
|
return true;
|
|
}
|
|
resp->set_error(errorMsg);
|
|
return false;
|
|
});
|
|
future.get();
|
|
return Status::OK;
|
|
}
|
|
|
|
Status GetGameStatus(ServerContext* ctx, const hydra::GameStatusRequest* req,
|
|
hydra::GameStatusResponse* resp) override {
|
|
// #1102: require a valid session token (metadata authorization).
|
|
// #1269: any authenticated session may learn running/up; host PIDs
|
|
// are admin-only (Start/Stop/Restart already require admin).
|
|
std::string sid = getSessionId(ctx, "");
|
|
auto future = workQueue_.enqueue<bool>(
|
|
[sid, gameName = req->game_name(), resp]
|
|
(SessionManager& sm, AccountManager& am, const HydraConfig& cfg, ProcessManager& pm) {
|
|
HydraSession* s = sid.empty() ? nullptr : sm.findByPersistId(sid);
|
|
if (!s) {
|
|
return false;
|
|
}
|
|
const bool admin = am.isAdmin(s->accountId);
|
|
for (const auto& g : cfg.games) {
|
|
if (g.type != GameType::Local) continue;
|
|
if (!gameName.empty() && g.name != gameName) continue;
|
|
auto* pi = resp->add_processes();
|
|
pi->set_game_name(g.name);
|
|
pi->set_running(pm.isRunning(g.name));
|
|
if (admin) {
|
|
pi->set_pid(static_cast<int>(pm.getPid(g.name)));
|
|
}
|
|
}
|
|
return true;
|
|
});
|
|
if (!future.get()) {
|
|
return Status(StatusCode::UNAUTHENTICATED, "session required");
|
|
}
|
|
return Status::OK;
|
|
}
|
|
|
|
private:
|
|
SessionManager& sessionMgr_;
|
|
AccountManager& accounts_;
|
|
const HydraConfig& config_;
|
|
ProcessManager& procMgr_;
|
|
WorkQueue& workQueue_;
|
|
};
|
|
|
|
// ---- GrpcServer lifecycle ----
|
|
|
|
GrpcServer::GrpcServer(SessionManager& sessionMgr, AccountManager& accounts,
|
|
const HydraConfig& config, ProcessManager& procMgr,
|
|
WorkQueue& workQueue)
|
|
: sessionMgr_(sessionMgr), accounts_(accounts), config_(config),
|
|
procMgr_(procMgr), workQueue_(workQueue) {
|
|
}
|
|
|
|
GrpcServer::~GrpcServer() {
|
|
shutdown();
|
|
}
|
|
|
|
bool GrpcServer::start(const std::string& listenAddr, std::string& errorMsg) {
|
|
auto service = std::make_unique<HydraServiceImpl>(
|
|
sessionMgr_, accounts_, config_, procMgr_, workQueue_);
|
|
|
|
ServerBuilder builder;
|
|
|
|
// Use TLS if cert/key are configured for gRPC
|
|
if (!config_.grpcTlsCert.empty() && !config_.grpcTlsKey.empty()) {
|
|
std::ifstream certFile(config_.grpcTlsCert);
|
|
std::ifstream keyFile(config_.grpcTlsKey);
|
|
if (!certFile.is_open() || !keyFile.is_open()) {
|
|
errorMsg = "cannot open gRPC TLS cert/key files";
|
|
return false;
|
|
}
|
|
std::string cert((std::istreambuf_iterator<char>(certFile)),
|
|
std::istreambuf_iterator<char>());
|
|
std::string key((std::istreambuf_iterator<char>(keyFile)),
|
|
std::istreambuf_iterator<char>());
|
|
|
|
grpc::SslServerCredentialsOptions sslOpts;
|
|
sslOpts.pem_key_cert_pairs.push_back({key, cert});
|
|
builder.AddListeningPort(listenAddr,
|
|
grpc::SslServerCredentials(sslOpts));
|
|
LOG_INFO("gRPC using TLS: cert=%s key=%s",
|
|
config_.grpcTlsCert.c_str(), config_.grpcTlsKey.c_str());
|
|
} else {
|
|
// Without TLS, only allow binding to loopback addresses
|
|
bool isLoopback = (listenAddr.find("127.0.0.1") != std::string::npos
|
|
|| listenAddr.find("[::1]") != std::string::npos
|
|
|| listenAddr.find("localhost") != std::string::npos);
|
|
if (!isLoopback) {
|
|
errorMsg = "gRPC without TLS must bind to loopback "
|
|
"(127.0.0.1 or [::1]), got: " + listenAddr;
|
|
return false;
|
|
}
|
|
builder.AddListeningPort(listenAddr, grpc::InsecureServerCredentials());
|
|
LOG_INFO("gRPC using insecure credentials on loopback: %s",
|
|
listenAddr.c_str());
|
|
}
|
|
|
|
builder.RegisterService(service.get());
|
|
|
|
server_ = builder.BuildAndStart();
|
|
if (!server_) {
|
|
errorMsg = "failed to start gRPC server on " + listenAddr;
|
|
return false;
|
|
}
|
|
|
|
LOG_INFO("gRPC server listening on %s", listenAddr.c_str());
|
|
|
|
serverThread_ = std::thread([srv = server_.get(), svc = std::move(service)]() {
|
|
srv->Wait();
|
|
});
|
|
|
|
return true;
|
|
}
|
|
|
|
void GrpcServer::shutdown() {
|
|
if (server_) {
|
|
server_->Shutdown();
|
|
if (serverThread_.joinable()) {
|
|
serverThread_.join();
|
|
}
|
|
server_.reset();
|
|
LOG_INFO("gRPC server shut down");
|
|
}
|
|
}
|
|
|
|
#endif // GRPC_ENABLED
|