Add coordinator login lock acquire request
This commit is contained in:
+336
-8
@@ -1,8 +1,12 @@
|
||||
#include "AccountSync.hh"
|
||||
|
||||
#include "AsyncUtils.hh"
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
#include <cstdio>
|
||||
#include <filesystem>
|
||||
#include <memory>
|
||||
#include <format>
|
||||
#include <fstream>
|
||||
#include <mutex>
|
||||
@@ -25,6 +29,239 @@ static Config get_config() {
|
||||
return current_config;
|
||||
}
|
||||
|
||||
struct ParsedHTTPURL {
|
||||
std::string host;
|
||||
uint16_t port = 80;
|
||||
std::string path = "/";
|
||||
};
|
||||
|
||||
static ParsedHTTPURL parse_http_url(const std::string& url) {
|
||||
static const std::string prefix = "http://";
|
||||
if (!url.starts_with(prefix)) {
|
||||
throw std::runtime_error("only http:// coordinator URLs are supported");
|
||||
}
|
||||
|
||||
size_t host_start = prefix.size();
|
||||
size_t path_start = url.find('/', host_start);
|
||||
std::string host_port = (path_start == std::string::npos)
|
||||
? url.substr(host_start)
|
||||
: url.substr(host_start, path_start - host_start);
|
||||
|
||||
ParsedHTTPURL ret;
|
||||
ret.path = (path_start == std::string::npos) ? "/" : url.substr(path_start);
|
||||
if (host_port.empty()) {
|
||||
throw std::runtime_error("coordinator URL has empty host");
|
||||
}
|
||||
|
||||
size_t colon_offset = host_port.rfind(':');
|
||||
if (colon_offset == std::string::npos) {
|
||||
ret.host = host_port;
|
||||
} else {
|
||||
ret.host = host_port.substr(0, colon_offset);
|
||||
std::string port_s = host_port.substr(colon_offset + 1);
|
||||
if (ret.host.empty() || port_s.empty()) {
|
||||
throw std::runtime_error("coordinator URL has invalid host/port");
|
||||
}
|
||||
size_t end_offset = 0;
|
||||
uint64_t port = std::stoull(port_s, &end_offset, 10);
|
||||
if ((end_offset != port_s.size()) || (port == 0) || (port > 0xFFFF)) {
|
||||
throw std::runtime_error("coordinator URL has invalid port");
|
||||
}
|
||||
ret.port = port;
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
static std::string join_url_path(std::string base_path, const std::string& suffix) {
|
||||
while ((base_path.size() > 1) && (base_path.back() == '/')) {
|
||||
base_path.pop_back();
|
||||
}
|
||||
if (base_path.empty() || (base_path == "/")) {
|
||||
return suffix;
|
||||
}
|
||||
return base_path + suffix;
|
||||
}
|
||||
|
||||
static std::string lowercase(std::string s) {
|
||||
std::transform(s.begin(), s.end(), s.begin(), [](unsigned char ch) -> char {
|
||||
return static_cast<char>(std::tolower(ch));
|
||||
});
|
||||
return s;
|
||||
}
|
||||
|
||||
static void trim_ascii_inplace(std::string& s) {
|
||||
while (!s.empty() && ((s.back() == ' ') || (s.back() == '\t') || (s.back() == '\r') || (s.back() == '\n'))) {
|
||||
s.pop_back();
|
||||
}
|
||||
size_t start = 0;
|
||||
while ((start < s.size()) && ((s[start] == ' ') || (s[start] == '\t') || (s[start] == '\r') || (s[start] == '\n'))) {
|
||||
start++;
|
||||
}
|
||||
if (start) {
|
||||
s = s.substr(start);
|
||||
}
|
||||
}
|
||||
|
||||
static asio::awaitable<std::string> read_http_line(
|
||||
asio::ip::tcp::socket& sock,
|
||||
std::string& pending_data,
|
||||
size_t max_length) {
|
||||
static const char* delimiter = "\r\n";
|
||||
static const size_t delimiter_size = 2;
|
||||
|
||||
size_t delimiter_pos = pending_data.find(delimiter);
|
||||
while ((delimiter_pos == std::string::npos) && (pending_data.size() < max_length)) {
|
||||
size_t pre_size = pending_data.size();
|
||||
pending_data.resize(std::min(max_length, pending_data.size() + 0x400));
|
||||
size_t bytes_read = co_await sock.async_read_some(
|
||||
asio::buffer(pending_data.data() + pre_size, pending_data.size() - pre_size),
|
||||
asio::use_awaitable);
|
||||
pending_data.resize(pre_size + bytes_read);
|
||||
delimiter_pos = pending_data.find(delimiter, (pre_size >= 1) ? (pre_size - 1) : 0);
|
||||
}
|
||||
|
||||
if (delimiter_pos == std::string::npos) {
|
||||
throw std::runtime_error("HTTP response line exceeds maximum length");
|
||||
}
|
||||
|
||||
std::string ret = pending_data.substr(0, delimiter_pos);
|
||||
pending_data = pending_data.substr(delimiter_pos + delimiter_size);
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
static asio::awaitable<std::string> read_http_data(
|
||||
asio::ip::tcp::socket& sock,
|
||||
std::string& pending_data,
|
||||
size_t size) {
|
||||
std::string ret;
|
||||
if (pending_data.size() == size) {
|
||||
pending_data.swap(ret);
|
||||
} else if (pending_data.size() > size) {
|
||||
ret = pending_data.substr(0, size);
|
||||
pending_data = pending_data.substr(size);
|
||||
} else {
|
||||
size_t bytes_to_read = size - pending_data.size();
|
||||
pending_data.swap(ret);
|
||||
ret.resize(size);
|
||||
co_await asio::async_read(
|
||||
sock,
|
||||
asio::buffer(ret.data() + size - bytes_to_read, bytes_to_read),
|
||||
asio::use_awaitable);
|
||||
}
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
static asio::awaitable<phosg::JSON> post_json_with_timeout(
|
||||
const Config& cfg,
|
||||
const std::string& path_suffix,
|
||||
const std::string& body) {
|
||||
ParsedHTTPURL url = parse_http_url(cfg.coordinator_url);
|
||||
std::string path = join_url_path(url.path, path_suffix);
|
||||
|
||||
auto executor = co_await asio::this_coro::executor;
|
||||
auto resolver = std::make_shared<asio::ip::tcp::resolver>(executor);
|
||||
auto sock = std::make_shared<asio::ip::tcp::socket>(executor);
|
||||
auto timer = std::make_shared<asio::steady_timer>(executor);
|
||||
auto timed_out = std::make_shared<bool>(false);
|
||||
|
||||
timer->expires_after(std::chrono::microseconds(cfg.request_timeout_usecs));
|
||||
timer->async_wait([resolver, sock, timed_out](std::error_code ec) -> void {
|
||||
if (!ec) {
|
||||
*timed_out = true;
|
||||
resolver->cancel();
|
||||
if (sock->is_open()) {
|
||||
sock->close();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
try {
|
||||
auto endpoints = co_await resolver->async_resolve(url.host, std::format("{}", url.port), asio::use_awaitable);
|
||||
co_await asio::async_connect(*sock, endpoints, asio::use_awaitable);
|
||||
|
||||
std::string host_header = url.host;
|
||||
if (url.port != 80) {
|
||||
host_header += std::format(":{}", url.port);
|
||||
}
|
||||
|
||||
std::string request = std::format(
|
||||
"POST {} HTTP/1.1\r\n"
|
||||
"Host: {}\r\n"
|
||||
"User-Agent: psopeeps-newserv\r\n"
|
||||
"Content-Type: application/json\r\n"
|
||||
"Accept: application/json\r\n"
|
||||
"Connection: close\r\n"
|
||||
"X-Psopeeps-Admin-Secret: {}\r\n"
|
||||
"Content-Length: {}\r\n"
|
||||
"\r\n"
|
||||
"{}",
|
||||
path,
|
||||
host_header,
|
||||
cfg.shared_secret,
|
||||
body.size(),
|
||||
body);
|
||||
|
||||
co_await asio::async_write(*sock, asio::buffer(request), asio::use_awaitable);
|
||||
|
||||
std::string pending_data;
|
||||
std::string status_line = co_await read_http_line(*sock, pending_data, 0x1000);
|
||||
if (!status_line.starts_with("HTTP/1.")) {
|
||||
throw std::runtime_error("invalid HTTP response from coordinator");
|
||||
}
|
||||
|
||||
size_t first_space = status_line.find(' ');
|
||||
if (first_space == std::string::npos) {
|
||||
throw std::runtime_error("invalid HTTP status line from coordinator");
|
||||
}
|
||||
size_t second_space = status_line.find(' ', first_space + 1);
|
||||
std::string code_s = status_line.substr(
|
||||
first_space + 1,
|
||||
(second_space == std::string::npos) ? std::string::npos : (second_space - first_space - 1));
|
||||
int response_code = std::stoi(code_s);
|
||||
|
||||
size_t content_length = 0;
|
||||
for (;;) {
|
||||
std::string line = co_await read_http_line(*sock, pending_data, 0x10000);
|
||||
if (line.empty()) {
|
||||
break;
|
||||
}
|
||||
size_t colon_offset = line.find(':');
|
||||
if (colon_offset == std::string::npos) {
|
||||
continue;
|
||||
}
|
||||
std::string name = lowercase(line.substr(0, colon_offset));
|
||||
std::string value = line.substr(colon_offset + 1);
|
||||
trim_ascii_inplace(value);
|
||||
if (name == "content-length") {
|
||||
size_t end_offset = 0;
|
||||
content_length = std::stoull(value, &end_offset, 10);
|
||||
if (end_offset != value.size()) {
|
||||
throw std::runtime_error("invalid Content-Length from coordinator");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (response_code != 200) {
|
||||
throw std::runtime_error(std::format("coordinator returned HTTP {}", response_code));
|
||||
}
|
||||
if (content_length > 0x100000) {
|
||||
throw std::runtime_error("coordinator response is too large");
|
||||
}
|
||||
|
||||
std::string response_body = co_await read_http_data(*sock, pending_data, content_length);
|
||||
timer->cancel();
|
||||
co_return phosg::JSON::parse(response_body);
|
||||
|
||||
} catch (...) {
|
||||
timer->cancel();
|
||||
if (*timed_out) {
|
||||
throw std::runtime_error("coordinator request timed out");
|
||||
}
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
static std::string source_label(const Config& cfg) {
|
||||
if (!cfg.source.empty()) {
|
||||
return cfg.source;
|
||||
@@ -186,15 +423,106 @@ asio::awaitable<LoginLockAcquireResult> acquire_login_lock(
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
ret.session_nonce = std::format("{}-{}-{}", source_label(cfg), account_id, now_usecs());
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] warning login_locks enabled but coordinator acquire is not implemented; allowing account_id=%010u source=%s version=%s nonce=%s\n",
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str(),
|
||||
ret.session_nonce.c_str());
|
||||
std::string proposed_session_nonce = std::format("{}-{}-{}", source_label(cfg), account_id, now_usecs());
|
||||
|
||||
co_return ret;
|
||||
if (cfg.coordinator_url.empty()) {
|
||||
std::string message = "account lock coordinator URL is not configured";
|
||||
if (cfg.fail_open) {
|
||||
ret.allowed = true;
|
||||
ret.fail_open_used = true;
|
||||
ret.session_nonce = proposed_session_nonce;
|
||||
ret.message = message;
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] warning login_lock_fail_open reason=%s account_id=%010u source=%s version=%s nonce=%s\n",
|
||||
message.c_str(),
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str(),
|
||||
ret.session_nonce.c_str());
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
ret.allowed = false;
|
||||
ret.message = "$C6Account lock server\nis unavailable.";
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] login_lock_denied reason=%s account_id=%010u source=%s version=%s\n",
|
||||
message.c_str(),
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str());
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
std::string body = std::format(
|
||||
"{{\"account_id\":{},\"account_id_str\":\"{:010}\",\"source\":\"{}\",\"source_region\":\"{}\",\"source_ship\":\"{}\",\"account_store\":\"{}\",\"version\":\"{}\",\"session_nonce\":\"{}\"}}",
|
||||
static_cast<unsigned int>(account_id),
|
||||
static_cast<unsigned int>(account_id),
|
||||
json_escape(source_label(cfg)),
|
||||
json_escape(cfg.source_region),
|
||||
json_escape(cfg.source_ship),
|
||||
json_escape(cfg.account_store),
|
||||
json_escape(version_name),
|
||||
json_escape(proposed_session_nonce));
|
||||
|
||||
try {
|
||||
phosg::JSON response = co_await post_json_with_timeout(cfg, "/account-locks/acquire", body);
|
||||
|
||||
ret.allowed = response.get_bool("ok", response.get_bool("OK", false));
|
||||
ret.session_nonce = response.get_string("session_nonce", proposed_session_nonce);
|
||||
ret.message = response.get_string("message", "");
|
||||
ret.holder_source = response.get_string("holder_source", "");
|
||||
|
||||
if (ret.allowed) {
|
||||
if (ret.session_nonce.empty()) {
|
||||
ret.session_nonce = proposed_session_nonce;
|
||||
}
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] login_lock_acquired account_id=%010u source=%s version=%s nonce=%s\n",
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str(),
|
||||
ret.session_nonce.c_str());
|
||||
} else {
|
||||
if (ret.message.empty()) {
|
||||
ret.message = "$C6Account is already active\non another ship.";
|
||||
}
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] login_lock_denied account_id=%010u source=%s version=%s holder_source=%s message=%s\n",
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str(),
|
||||
ret.holder_source.c_str(),
|
||||
ret.message.c_str());
|
||||
}
|
||||
|
||||
co_return ret;
|
||||
|
||||
} catch (const std::exception& e) {
|
||||
if (cfg.fail_open) {
|
||||
ret.allowed = true;
|
||||
ret.fail_open_used = true;
|
||||
ret.session_nonce = proposed_session_nonce;
|
||||
ret.message = e.what();
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] warning login_lock_fail_open reason=%s account_id=%010u source=%s version=%s nonce=%s\n",
|
||||
e.what(),
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str(),
|
||||
ret.session_nonce.c_str());
|
||||
co_return ret;
|
||||
}
|
||||
|
||||
ret.allowed = false;
|
||||
ret.message = "$C6Account lock server\nis unavailable.";
|
||||
std::fprintf(stderr,
|
||||
"[AccountSync] login_lock_denied reason=%s account_id=%010u source=%s version=%s\n",
|
||||
e.what(),
|
||||
static_cast<unsigned int>(account_id),
|
||||
source_label(cfg).c_str(),
|
||||
version_name.c_str());
|
||||
co_return ret;
|
||||
}
|
||||
}
|
||||
|
||||
void notify_login_session_end(
|
||||
|
||||
Reference in New Issue
Block a user