mirror of
https://github.com/kunkundi/crossdesk.git
synced 2026-10-28 00:33:06 +08:00
807 lines
29 KiB
C++
807 lines
29 KiB
C++
#include "lan_session_manager.h"
|
|
|
|
#include <openssl/crypto.h>
|
|
|
|
#include <algorithm>
|
|
#include <thread>
|
|
|
|
namespace crossdesk::lan {
|
|
namespace {
|
|
using Clock = std::chrono::steady_clock;
|
|
using Json = nlohmann::json;
|
|
template <class T>
|
|
bool Ready(std::future<T>& future) {
|
|
return future.valid() &&
|
|
future.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
|
|
}
|
|
void Check(bool condition) {
|
|
if (!condition) throw std::runtime_error("Invalid LAN protocol message");
|
|
}
|
|
bool Id(const Json& value) {
|
|
if (!value.is_string()) return false;
|
|
const auto& s = value.get_ref<const std::string&>();
|
|
return s.size() == 32 && std::all_of(s.begin(), s.end(), [](char c) {
|
|
return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f');
|
|
});
|
|
}
|
|
Json Capabilities() {
|
|
return Json::array({"host-only-ipv4", "dtls-srtp-all", "password-auth-v1"});
|
|
}
|
|
void Capable(const Json& j) {
|
|
const auto& caps = j.at("capabilities");
|
|
Check(caps.is_array() && caps.size() <= 16);
|
|
for (const auto& required : Capabilities())
|
|
Check(std::find(caps.begin(), caps.end(), required) != caps.end());
|
|
}
|
|
void Wipe(std::string& text) {
|
|
if (!text.empty()) OPENSSL_cleanse(text.data(), text.size());
|
|
text.clear();
|
|
}
|
|
const char* FailureReason(const Session& s, const char* fallback) {
|
|
switch (s.failure_code.load()) {
|
|
case MiniRtcDirectInvalidSignal:
|
|
return "InvalidSignal";
|
|
case MiniRtcDirectSignalSendFailed:
|
|
return "SignalSendFailed";
|
|
case MiniRtcDirectDtlsHandshakeFailed:
|
|
return "DtlsHandshakeFailed";
|
|
case MiniRtcDirectRequiredCryptoUnsupported:
|
|
return "RequiredCryptoUnsupported";
|
|
case MiniRtcDirectSrtpInstallFailed:
|
|
return "SrtpInstallFailed";
|
|
case MiniRtcDirectNetworkPolicyFailed:
|
|
return "NetworkPolicyFailed";
|
|
case MiniRtcDirectFingerprintMissing:
|
|
return "FingerprintMissing";
|
|
case MiniRtcDirectFingerprintMismatch:
|
|
return "FingerprintMismatch";
|
|
case MiniRtcDirectSrtpKeyExportFailed:
|
|
return "SrtpKeyExportFailed";
|
|
default:
|
|
return fallback;
|
|
}
|
|
}
|
|
} // namespace
|
|
int Session::SendSignal(const char* data, size_t size, void* context) {
|
|
auto* s = static_cast<Session*>(context);
|
|
if (!s || s->closing || !data || !size || size > 256 * 1024) return -1;
|
|
std::lock_guard lock(s->signal_mutex);
|
|
if (s->closing || s->signal_count >= 256 ||
|
|
s->signal_bytes + size > 1024 * 1024)
|
|
return -1;
|
|
++s->signal_count;
|
|
s->signal_bytes += size;
|
|
s->signals.emplace_back(data, size);
|
|
return 0;
|
|
}
|
|
void Session::Error(MiniRtcDirectError error, void* context) {
|
|
if (auto* s = static_cast<Session*>(context)) {
|
|
int empty = -1;
|
|
s->failure_code.compare_exchange_strong(empty, error);
|
|
s->failed = true;
|
|
}
|
|
}
|
|
void Session::Status(ConnectionStatus status) {
|
|
if (closing) return;
|
|
if (status == ConnectionStatus::Connected)
|
|
ready = true;
|
|
else if (status == ConnectionStatus::Closed ||
|
|
status == ConnectionStatus::Failed ||
|
|
status == ConnectionStatus::Disconnected)
|
|
failed = true;
|
|
}
|
|
SessionManager::SessionManager(std::filesystem::path path, Hooks hooks)
|
|
: identity_(std::move(path)), hooks_(std::move(hooks)) {
|
|
identity_.Load();
|
|
access_password_ = identity_.ReadPassword();
|
|
new_password_ = access_password_;
|
|
if (!Identity::CurrentVerifier(identity_.verifier())) {
|
|
if (!access_password_.empty()) {
|
|
// Migrate the saved password, retaining its value, identity and settings.
|
|
SetPassword(access_password_);
|
|
} else if (identity_.verifier().is_null()) {
|
|
// Generate the initial password before automatically accepting connections.
|
|
ResetPassword();
|
|
} else {
|
|
hooks_.error("LAN password is not ready");
|
|
}
|
|
}
|
|
}
|
|
SessionManager::~SessionManager() {
|
|
Stop();
|
|
while (!Quiescent()) {
|
|
Tick();
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(2));
|
|
}
|
|
transport_.Shutdown();
|
|
Wipe(new_password_);
|
|
Wipe(access_password_);
|
|
}
|
|
void SessionManager::Configure(const std::string& address, unsigned port) {
|
|
if (stopping_) return;
|
|
if (!port || port > 65535) throw std::runtime_error("Invalid LAN port");
|
|
if (!Identity::CurrentVerifier(identity_.verifier()))
|
|
throw std::runtime_error("LAN password is not ready");
|
|
ReplaceListener(address, port, true);
|
|
}
|
|
void SessionManager::ReplaceListener(const std::string& address, unsigned port,
|
|
bool save, bool allow_fallback) {
|
|
const auto previous = listeners_.find(address);
|
|
if (previous != listeners_.end() && previous->second.port == port) return;
|
|
const auto next = [&] {
|
|
try {
|
|
return transport_.Listen(identity_, address, port);
|
|
} catch (const std::exception& e) {
|
|
const std::string error = e.what();
|
|
if (!allow_fallback ||
|
|
(error != "LanPortInUse" && error != "LanPortPermissionDenied" &&
|
|
error != "LanPortCheckFailed"))
|
|
throw;
|
|
// Keep the OS-assigned socket bound through persistence/publication;
|
|
// probing then releasing a free port would introduce an allocation race.
|
|
return transport_.Listen(identity_, address, 0);
|
|
}
|
|
}();
|
|
try {
|
|
if (save) identity_.SaveSettings(address, next.port);
|
|
} catch (...) {
|
|
transport_.StopAccept(next.id);
|
|
throw;
|
|
}
|
|
const auto old = previous == listeners_.end() ? 0 : previous->second.id;
|
|
listeners_[address] = {next.id, next.port};
|
|
SetListenerError(address, next.port, "");
|
|
if (old) RetireListener(old);
|
|
}
|
|
void SessionManager::RetireListener(uint64_t id) {
|
|
transport_.StopAccept(id);
|
|
// Only cancel unfinished authentication on this retired endpoint. Other
|
|
// addresses and sessions that already authenticated keep their channels.
|
|
for (auto& [_, s] : sessions_)
|
|
if (s->incoming && s->channel->listener == id &&
|
|
s->phase < Session::AwaitOffer)
|
|
Close(s, "ListenerChanged");
|
|
}
|
|
void SessionManager::ResetPassword() {
|
|
if (stopping_ || reset_job_.valid()) return;
|
|
reset_job_ =
|
|
std::async(std::launch::async, [current = access_password_]() mutable {
|
|
try {
|
|
for (;;) {
|
|
auto result = Identity::GeneratePassword();
|
|
if (result.first != current) {
|
|
Wipe(current);
|
|
return result;
|
|
}
|
|
Wipe(result.first);
|
|
}
|
|
} catch (...) {
|
|
Wipe(current);
|
|
throw;
|
|
}
|
|
});
|
|
}
|
|
void SessionManager::SetPassword(std::string password) {
|
|
if (!ValidPassword(password)) {
|
|
Wipe(password);
|
|
throw std::runtime_error(
|
|
"LAN password must contain exactly 6 ASCII letters or digits");
|
|
}
|
|
if (stopping_ || reset_job_.valid() ||
|
|
(password == access_password_ &&
|
|
Identity::CurrentVerifier(identity_.verifier()))) {
|
|
Wipe(password);
|
|
return;
|
|
}
|
|
reset_job_ = std::async(
|
|
std::launch::async, [password = std::move(password)]() mutable {
|
|
try {
|
|
auto record = Identity::RegisterPassword(password);
|
|
return std::make_pair(std::move(password), std::move(record));
|
|
} catch (...) {
|
|
Wipe(password);
|
|
throw;
|
|
}
|
|
});
|
|
}
|
|
void SessionManager::SetListenerError(const std::string& address, unsigned port,
|
|
const std::string& error) {
|
|
if (error.empty()) {
|
|
listener_errors_.erase(address);
|
|
listener_warnings_.erase(address);
|
|
return;
|
|
}
|
|
const auto previous = listener_errors_.find(address);
|
|
if (previous != listener_errors_.end() && previous->second.port == port &&
|
|
previous->second.error == error)
|
|
return;
|
|
ListenerIssue issue{address, port, error};
|
|
listener_errors_[address] = issue;
|
|
listener_warnings_[address] = std::move(issue);
|
|
}
|
|
std::string SessionManager::ListenerError(const std::string& address) const {
|
|
const auto it = listener_errors_.find(address);
|
|
return it == listener_errors_.end() ? "" : it->second.error;
|
|
}
|
|
std::optional<SessionManager::ListenerIssue> SessionManager::TakeListenerWarning() {
|
|
if (listener_warnings_.empty()) return std::nullopt;
|
|
auto it = listener_warnings_.begin();
|
|
auto issue = std::move(it->second);
|
|
listener_warnings_.erase(it);
|
|
return issue;
|
|
}
|
|
void SessionManager::RefreshListeners() {
|
|
if (stopping_ || !Identity::CurrentVerifier(identity_.verifier())) return;
|
|
const auto available = [&](const auto& address) {
|
|
return std::any_of(interfaces_.begin(), interfaces_.end(),
|
|
[&](const auto& i) { return i.address == address; });
|
|
};
|
|
for (auto it = listeners_.begin(); it != listeners_.end();) {
|
|
if (available(it->first)) {
|
|
++it;
|
|
continue;
|
|
}
|
|
RetireListener(it->second.id);
|
|
it = listeners_.erase(it);
|
|
}
|
|
for (auto it = listener_errors_.begin(); it != listener_errors_.end();) {
|
|
if (available(it->first)) {
|
|
++it;
|
|
continue;
|
|
}
|
|
listener_warnings_.erase(it->first);
|
|
it = listener_errors_.erase(it);
|
|
}
|
|
for (const auto& interface : interfaces_) {
|
|
const auto port = identity_.PortFor(interface.address);
|
|
const bool initialize = !identity_.HasPortFor(interface.address);
|
|
try {
|
|
// Persist the first choice even when the default works. Later conflicts
|
|
// must not silently change an endpoint users may already have shared.
|
|
ReplaceListener(interface.address, port, initialize, initialize);
|
|
} catch (const std::exception& e) {
|
|
SetListenerError(interface.address, port, e.what());
|
|
}
|
|
}
|
|
}
|
|
std::string SessionManager::TakeNewPassword() {
|
|
return std::exchange(new_password_, {});
|
|
}
|
|
SessionManager::SessionPtr SessionManager::Connect(const std::string& address,
|
|
unsigned port,
|
|
std::string password) {
|
|
if (stopping_ || sessions_.size() >= 24)
|
|
throw std::runtime_error("LAN connection limit reached");
|
|
if (!ValidPassword(password))
|
|
throw std::runtime_error(
|
|
"LAN password must contain exactly 6 ASCII letters or digits");
|
|
if (RetryAfter(address, port) > 0) {
|
|
Wipe(password);
|
|
throw std::runtime_error("AuthenticationRateLimited");
|
|
}
|
|
auto s = std::make_shared<Session>();
|
|
const auto [ip, endpoint_port] = ParseEndpoint(address, port);
|
|
s->target = Target{ip, {}, endpoint_port};
|
|
s->channel = transport_.Connect(s->target);
|
|
s->password = std::move(password);
|
|
s->epoch = epoch_;
|
|
s->key = "lan/" + RandomHex(16);
|
|
s->deadline = Clock::now() + std::chrono::seconds(8);
|
|
sessions_[s->channel->id] = s;
|
|
return s;
|
|
}
|
|
bool SessionManager::Send(const SessionPtr& s, Json j, bool sequenced) {
|
|
j["version"] = 2;
|
|
if (sequenced) {
|
|
j["session_id"] = s->session_id;
|
|
j["seq"] = s->send_seq + 1;
|
|
}
|
|
if (s->closing || !transport_.Send(s->channel, j.dump())) {
|
|
Close(s, "SignalSendFailed");
|
|
return false;
|
|
}
|
|
if (sequenced) ++s->send_seq;
|
|
return true;
|
|
}
|
|
int SessionManager::RetryAfter(const std::string& address,
|
|
unsigned port) const {
|
|
const auto [ip, endpoint_port] = ParseEndpoint(address, port);
|
|
const auto it = retry_at_.find(ip + ":" + std::to_string(endpoint_port));
|
|
const auto now = Clock::now();
|
|
return it == retry_at_.end() || it->second <= now
|
|
? 0
|
|
: static_cast<int>(
|
|
std::chrono::ceil<std::chrono::seconds>(it->second - now)
|
|
.count());
|
|
}
|
|
void SessionManager::Fail(const SessionPtr& s, const std::string& error,
|
|
int retry_after) {
|
|
Json reply = {{"type", "error"}, {"error", error}};
|
|
// Use the same failure codes and retry_after semantics as cloud Join.
|
|
if (error == "InvalidCredentials")
|
|
reply["error_code"] = "AUTHENTICATION_FAILED";
|
|
if (error == "AuthenticationRateLimited")
|
|
reply["error_code"] = "AUTHENTICATION_THROTTLED";
|
|
if (error == "Busy") reply["error_code"] = "CREDENTIAL_SERVICE_BUSY";
|
|
if (retry_after > 0) reply["retry_after"] = retry_after;
|
|
if (s->session_id.empty() ||
|
|
(s->incoming && s->phase < Session::AwaitOffer)) {
|
|
reply["request_id"] = s->auth_request;
|
|
Send(s, std::move(reply));
|
|
} else {
|
|
Send(s, std::move(reply), true);
|
|
}
|
|
Close(s, error);
|
|
}
|
|
void SessionManager::Prepare(const SessionPtr& s) {
|
|
size_t admitted = 0;
|
|
for (auto& [_, other] : sessions_)
|
|
if (!other->closing && (other->phase == Session::Preparing ||
|
|
other->phase >= Session::AwaitOffer))
|
|
++admitted;
|
|
if (admitted >= 8 || prepare_jobs_.size() >= 2) {
|
|
Fail(s, "Busy");
|
|
return;
|
|
}
|
|
s->phase = Session::Preparing;
|
|
if (s->incoming) {
|
|
s->session_id = RandomHex(16);
|
|
s->local_id = identity_.id();
|
|
s->remote_id = RandomHex(16);
|
|
}
|
|
auto work = hooks_.prepare(s);
|
|
prepare_jobs_.push_back(
|
|
{s, std::async(std::launch::async, [s, work = std::move(work)] {
|
|
if (s->closing) return Prepared{};
|
|
Prepared prepared = work();
|
|
if (!prepared.peer || s->closing) return prepared;
|
|
MiniRtcDirectOptions options{};
|
|
options.struct_size = sizeof(options);
|
|
options.version = 1;
|
|
options.local_id = s->local_id.c_str();
|
|
options.remote_id = s->remote_id.c_str();
|
|
options.session_id = s->session_id.c_str();
|
|
options.local_address = s->channel->local_address.c_str();
|
|
options.remote_address = s->channel->remote_address.c_str();
|
|
options.offerer = !s->incoming;
|
|
options.send_signal = Session::SendSignal;
|
|
options.on_error = Session::Error;
|
|
options.context = s.get();
|
|
if (InitDirectPeer(prepared.peer, &options) != 0 ||
|
|
(!s->closing &&
|
|
StartDirectSession(prepared.peer, !s->incoming) != 0))
|
|
s->failed = true;
|
|
return prepared;
|
|
})});
|
|
}
|
|
void SessionManager::Message(const SessionPtr& s, const std::string& text) {
|
|
auto j = Json::parse(text);
|
|
Check(j.is_object() && j.at("version").is_number_integer() &&
|
|
j.at("version") == 2 && j.at("type").is_string() &&
|
|
!j.contains("user_id") && !j.contains("remote_user_id"));
|
|
const auto type = j.at("type").get<std::string>();
|
|
if (type == "error" && (s->session_id.empty() ||
|
|
(s->incoming && s->phase < Session::AwaitOffer))) {
|
|
Check(!j.contains("session_id") && !j.contains("seq") &&
|
|
j.at("request_id") == s->auth_request);
|
|
const auto error = j.at("error").get<std::string>();
|
|
Check(error.size() <= 80);
|
|
if (!s->incoming &&
|
|
(error == "AuthenticationRateLimited" || error == "Busy") &&
|
|
j.contains("retry_after")) {
|
|
Check(j["retry_after"].is_number_integer() && j["retry_after"] > 0 &&
|
|
j["retry_after"] <= 900);
|
|
const auto now = Clock::now();
|
|
for (auto it = retry_at_.begin(); it != retry_at_.end();)
|
|
if (it->second <= now)
|
|
it = retry_at_.erase(it);
|
|
else
|
|
++it;
|
|
if (retry_at_.size() < 64)
|
|
retry_at_[s->target.address + ":" + std::to_string(s->target.port)] =
|
|
now + std::chrono::seconds(j["retry_after"].get<int>());
|
|
}
|
|
Close(s, error);
|
|
return;
|
|
}
|
|
if (s->phase == Session::Preparing && !s->incoming) {
|
|
Check(j.at("session_id") == s->session_id &&
|
|
j.at("seq").is_number_unsigned() &&
|
|
j.at("seq").get<uint64_t>() ==
|
|
s->receive_seq + s->pending_receive.size() + 1 &&
|
|
!j.contains("request_id"));
|
|
// Identity is already bound by auth_result. Process cancellation without
|
|
// waiting for Peer preparation, before the following WSS close can erase
|
|
// its reason; earlier queued SDP/candidates never grant authorization.
|
|
if (type == "bye" || type == "error") {
|
|
const auto reason = type == "bye" ? std::string("RemoteClosed")
|
|
: j.at("error").get<std::string>();
|
|
Check(!reason.empty() && reason.size() <= 80);
|
|
Close(s, reason);
|
|
return;
|
|
}
|
|
Check(type == "answer" || type == "candidate");
|
|
Check(s->pending_receive.size() < 128);
|
|
s->pending_receive.push_back(text);
|
|
return;
|
|
}
|
|
if (s->phase == Session::AwaitAuth) {
|
|
Check(type == "auth" && !j.contains("session_id") && !j.contains("seq") &&
|
|
Id(j.at("request_id")) && j.at("password").is_string() &&
|
|
!j.contains("share") && !j.contains("proof"));
|
|
Capable(j);
|
|
s->auth_request = j.at("request_id");
|
|
auto& submitted = j["password"].get_ref<std::string&>();
|
|
if (!ValidPassword(submitted)) {
|
|
Wipe(submitted);
|
|
Fail(s, "InvalidCredentials");
|
|
return;
|
|
}
|
|
if (auth_jobs_.size() >= 2) {
|
|
Wipe(submitted);
|
|
Fail(s, "Busy", 1);
|
|
return;
|
|
}
|
|
const auto delay =
|
|
credential_work_limiter_.Admit(s->channel->remote_address);
|
|
if (delay > 0) {
|
|
Wipe(submitted);
|
|
Fail(s, "AuthenticationRateLimited", delay);
|
|
return;
|
|
}
|
|
s->phase = Session::Authenticating;
|
|
auth_jobs_.push_back(
|
|
{s,
|
|
std::async(std::launch::async,
|
|
[this, password = std::exchange(submitted, {}),
|
|
record = identity_.verifier(),
|
|
source = s->channel->remote_address,
|
|
target = identity_.id()]() mutable {
|
|
try {
|
|
auto result = auth_limiter_.Verify(source, target, [&] {
|
|
return Identity::VerifyPassword(record, password);
|
|
});
|
|
Wipe(password);
|
|
return result;
|
|
} catch (...) {
|
|
Wipe(password);
|
|
throw;
|
|
}
|
|
})});
|
|
return;
|
|
}
|
|
if (s->phase == Session::AwaitAuthResult) {
|
|
Check(type == "auth_result" && j.at("request_id") == s->auth_request &&
|
|
!j.contains("seq"));
|
|
if (!j.value("success", false)) {
|
|
Close(s, "InvalidCredentials");
|
|
return;
|
|
}
|
|
Check(Id(j.at("session_id")) && Id(j.at("host_peer_id")) &&
|
|
Id(j.at("controller_peer_id")) &&
|
|
j.at("controller_peer_id") != j.at("host_peer_id"));
|
|
s->target.device_id = j.at("host_peer_id");
|
|
s->session_id = j.at("session_id");
|
|
s->local_id = j.at("controller_peer_id");
|
|
s->remote_id = j.at("host_peer_id");
|
|
Prepare(s);
|
|
return;
|
|
}
|
|
Check(s->peer && j.at("session_id") == s->session_id &&
|
|
j.at("seq").is_number_unsigned() &&
|
|
j.at("seq").get<uint64_t>() == s->receive_seq + 1 &&
|
|
!j.contains("request_id"));
|
|
++s->receive_seq;
|
|
if (type == "bye" || type == "error") {
|
|
const auto reason = type == "bye" ? std::string("RemoteClosed")
|
|
: j.at("error").get<std::string>();
|
|
Check(!reason.empty() && reason.size() <= 80);
|
|
Close(s, reason);
|
|
return;
|
|
}
|
|
if (type == "offer") {
|
|
Check(s->incoming && s->phase == Session::AwaitOffer);
|
|
s->phase = Session::Negotiating;
|
|
s->deadline = Clock::now() + std::chrono::seconds(15);
|
|
} else if (type == "answer") {
|
|
Check(!s->incoming && s->phase == Session::AwaitAnswer);
|
|
s->phase = Session::Negotiating;
|
|
} else
|
|
Check(type == "candidate" &&
|
|
(s->phase == Session::AwaitAnswer ||
|
|
s->phase == Session::Negotiating || s->phase == Session::Connected));
|
|
if (ReceiveDirectSignal(s->peer, text.data(), text.size()) != 0)
|
|
Fail(s, "InvalidSignal");
|
|
}
|
|
void SessionManager::Dispose(const SessionPtr& s) {
|
|
if (!s->peer) return;
|
|
auto* peer = std::exchange(s->peer, nullptr);
|
|
auto owner = s->callback_owner;
|
|
auto drain = s->drain_callbacks;
|
|
cleanup_jobs_.push_back(
|
|
std::async(std::launch::async, [s, peer, owner, drain]() mutable {
|
|
if (drain) drain();
|
|
CloseDirectSession(peer);
|
|
DestroyPeer(&peer);
|
|
}));
|
|
}
|
|
void SessionManager::Close(const SessionPtr& s, const std::string& error) {
|
|
if (!s || s->closing.exchange(true)) return;
|
|
s->authorized = false;
|
|
if (error.empty() && !s->session_id.empty() &&
|
|
s->phase >= Session::AwaitOffer)
|
|
transport_.Send(s->channel, Json{{"version", 2},
|
|
{"type", "bye"},
|
|
{"session_id", s->session_id},
|
|
{"seq", ++s->send_seq}}
|
|
.dump());
|
|
s->phase = Session::Closing;
|
|
s->error = error;
|
|
Wipe(s->password);
|
|
transport_.Close(s->channel);
|
|
hooks_.closed(s);
|
|
Dispose(s);
|
|
if (!error.empty() && !s->incoming && !stopping_) hooks_.error(error);
|
|
}
|
|
void SessionManager::Stop() {
|
|
if (stopping_) return;
|
|
stopping_ = true;
|
|
for (const auto& [_, listener] : listeners_) transport_.StopAccept(listener.id);
|
|
listeners_.clear();
|
|
listener_errors_.clear();
|
|
listener_warnings_.clear();
|
|
for (auto& [_, s] : sessions_) Close(s);
|
|
}
|
|
bool SessionManager::Quiescent() const {
|
|
return (!stopping_ || network_drained_) && sessions_.empty() &&
|
|
auth_jobs_.empty() && prepare_jobs_.empty() && cleanup_jobs_.empty() &&
|
|
!reset_job_.valid();
|
|
}
|
|
std::string SessionManager::PendingWork() const {
|
|
return "sessions=" + std::to_string(sessions_.size()) +
|
|
", auth=" + std::to_string(auth_jobs_.size()) +
|
|
", preparing=" + std::to_string(prepare_jobs_.size()) +
|
|
", cleanup=" + std::to_string(cleanup_jobs_.size()) +
|
|
", reset=" + std::to_string(reset_job_.valid()) +
|
|
", network_drained=" + std::to_string(network_drained_);
|
|
}
|
|
void SessionManager::Tick() {
|
|
if (Ready(reset_job_)) {
|
|
try {
|
|
auto [password, verifier] = reset_job_.get();
|
|
if (!stopping_) {
|
|
const bool changed = verifier != identity_.verifier();
|
|
try {
|
|
identity_.SetPassword(password, verifier);
|
|
} catch (...) {
|
|
Wipe(password);
|
|
throw;
|
|
}
|
|
if (changed) ++epoch_;
|
|
Wipe(new_password_);
|
|
Wipe(access_password_);
|
|
access_password_ = password;
|
|
new_password_ = std::move(password);
|
|
for (auto& [_, s] : sessions_)
|
|
if (s->incoming && !s->authorized && s->epoch != epoch_)
|
|
Fail(s, "CredentialsChanged");
|
|
interface_check_ = {}; // Start listening as soon as credentials are ready.
|
|
} else
|
|
Wipe(password);
|
|
} catch (const std::exception& e) {
|
|
hooks_.error(e.what());
|
|
}
|
|
}
|
|
for (auto it = auth_jobs_.begin(); it != auth_jobs_.end();) {
|
|
if (!Ready(it->result)) {
|
|
++it;
|
|
continue;
|
|
}
|
|
auto s = it->session;
|
|
DeviceAuthLimiter::Result result;
|
|
bool failed = false;
|
|
try {
|
|
result = it->result.get();
|
|
} catch (...) {
|
|
failed = true;
|
|
}
|
|
it = auth_jobs_.erase(it);
|
|
if (s->closing || stopping_ || s->epoch != epoch_) continue;
|
|
if (failed) {
|
|
Fail(s, "Busy", 1);
|
|
} else if (result.retry_after > 0) {
|
|
Fail(s, "AuthenticationRateLimited", result.retry_after);
|
|
} else if (!result.authenticated) {
|
|
Fail(s, "InvalidCredentials");
|
|
} else {
|
|
try {
|
|
Prepare(s);
|
|
} catch (...) {
|
|
Fail(s, "PeerPreparationFailed");
|
|
}
|
|
}
|
|
}
|
|
for (auto it = prepare_jobs_.begin(); it != prepare_jobs_.end();) {
|
|
if (!Ready(it->result)) {
|
|
++it;
|
|
continue;
|
|
}
|
|
auto s = it->session;
|
|
try {
|
|
auto p = it->result.get();
|
|
s->peer = p.peer;
|
|
s->callback_owner = std::move(p.callback_owner);
|
|
s->drain_callbacks = std::move(p.drain_callbacks);
|
|
} catch (...) {
|
|
s->failed = true;
|
|
}
|
|
it = prepare_jobs_.erase(it);
|
|
if (s->closing || stopping_ || (s->incoming && s->epoch != epoch_)) {
|
|
Dispose(s);
|
|
continue;
|
|
}
|
|
if (!s->peer || s->failed) {
|
|
Fail(s, FailureReason(*s, "PeerPreparationFailed"));
|
|
continue;
|
|
}
|
|
hooks_.attached(s);
|
|
transport_.Authenticated(s->channel);
|
|
if (s->incoming) {
|
|
s->phase = Session::AwaitOffer;
|
|
s->deadline = Clock::now() + std::chrono::seconds(5);
|
|
Send(s, {{"type", "auth_result"},
|
|
{"request_id", s->auth_request},
|
|
{"success", true},
|
|
{"session_id", s->session_id},
|
|
{"host_peer_id", s->local_id},
|
|
{"controller_peer_id", s->remote_id}});
|
|
} else {
|
|
s->phase = Session::AwaitAnswer;
|
|
s->deadline = Clock::now() + std::chrono::seconds(15);
|
|
while (!s->pending_receive.empty() && !s->closing) {
|
|
auto text = std::move(s->pending_receive.front());
|
|
s->pending_receive.pop_front();
|
|
try {
|
|
Message(s, text);
|
|
} catch (...) {
|
|
Fail(s, "InvalidSignal");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for (auto& event : transport_.Poll()) {
|
|
auto it = sessions_.find(event.channel->id);
|
|
if (it == sessions_.end()) {
|
|
const auto listener = listeners_.find(event.channel->local_address);
|
|
if (event.kind != Transport::Event::Open || !event.channel->incoming ||
|
|
stopping_ || listener == listeners_.end() ||
|
|
event.channel->listener != listener->second.id) {
|
|
transport_.Close(event.channel);
|
|
continue;
|
|
}
|
|
auto s = std::make_shared<Session>();
|
|
s->channel = event.channel;
|
|
s->incoming = true;
|
|
s->key = "lan/" + RandomHex(16);
|
|
s->epoch = epoch_;
|
|
sessions_[event.channel->id] = s;
|
|
it = sessions_.find(event.channel->id);
|
|
}
|
|
auto s = it->second;
|
|
if (s->closing) continue;
|
|
if (event.kind == Transport::Event::Open) {
|
|
s->deadline = Clock::now() + std::chrono::seconds(10);
|
|
s->phase = s->incoming ? Session::AwaitAuth : Session::AwaitAuthResult;
|
|
if (!s->incoming) {
|
|
s->auth_request = RandomHex(16);
|
|
Json auth = {{"type", "auth"},
|
|
{"request_id", s->auth_request},
|
|
{"password", std::exchange(s->password, {})},
|
|
{"capabilities", Capabilities()}};
|
|
Send(s, auth);
|
|
Wipe(auth["password"].get_ref<std::string&>());
|
|
}
|
|
} else if (event.kind == Transport::Event::Closed)
|
|
Close(s, event.payload.empty() ? "ConnectionClosed" : event.payload);
|
|
else
|
|
try {
|
|
Message(s, event.payload);
|
|
} catch (...) {
|
|
Fail(s, "InvalidSignal");
|
|
}
|
|
Wipe(event.payload);
|
|
}
|
|
const auto now = Clock::now();
|
|
for (auto& [_, s] : sessions_) {
|
|
if (s->closing) continue;
|
|
if (s->failed) {
|
|
Fail(s, FailureReason(*s, "SecureTransportFailed"));
|
|
continue;
|
|
}
|
|
if (s->phase != Session::Connected && now >= s->deadline) {
|
|
Fail(s, s->phase == Session::Negotiating ? "IceConnectionTimeout"
|
|
: "Timeout");
|
|
continue;
|
|
}
|
|
std::deque<std::string> signals;
|
|
{
|
|
std::lock_guard lock(s->signal_mutex);
|
|
signals.swap(s->signals);
|
|
}
|
|
for (const auto& raw : signals) try {
|
|
auto j = Json::parse(raw);
|
|
Check(j.at("transmission_id") == s->session_id &&
|
|
j.at("user_id") == s->local_id &&
|
|
j.at("remote_user_id") == s->remote_id);
|
|
const auto type = j.at("type").get<std::string>();
|
|
Json out;
|
|
if (type == "offer" || type == "answer")
|
|
out = {{"type", type}, {"sdp", j.at("sdp")}};
|
|
else {
|
|
Check(type == "new_candidate");
|
|
out = {{"type", "candidate"}, {"ufrag", j.at("ufrag")}};
|
|
if (j.at("sdp") == "a=end-of-candidates\r\n")
|
|
out["end_of_candidates"] = true;
|
|
else
|
|
out["candidate"] = j.at("sdp");
|
|
}
|
|
if (!Send(s, std::move(out), true)) break;
|
|
} catch (...) {
|
|
Fail(s, "InvalidLocalSignal");
|
|
break;
|
|
}
|
|
if (s->ready.exchange(false) && !s->closing && !s->failed) {
|
|
if (s->phase != Session::Negotiating || stopping_ ||
|
|
(s->incoming && s->epoch != epoch_)) {
|
|
Fail(s, "CredentialsChanged");
|
|
continue;
|
|
}
|
|
s->phase = Session::Connected;
|
|
s->authorized = true;
|
|
hooks_.connected(s);
|
|
}
|
|
}
|
|
if (!stopping_ && now - interface_check_ >= std::chrono::seconds(2)) {
|
|
interface_check_ = now;
|
|
interfaces_ = PhysicalInterfaces();
|
|
RefreshListeners();
|
|
for (auto& [_, s] : sessions_)
|
|
if (s->phase != Session::Connecting &&
|
|
!s->channel->local_address.empty() &&
|
|
std::none_of(interfaces_.begin(), interfaces_.end(),
|
|
[&](const auto& i) {
|
|
return i.address == s->channel->local_address;
|
|
}))
|
|
Close(s, "InterfaceRemoved");
|
|
}
|
|
for (auto it = cleanup_jobs_.begin(); it != cleanup_jobs_.end();)
|
|
if (Ready(*it)) {
|
|
try {
|
|
it->get();
|
|
} catch (...) {
|
|
}
|
|
it = cleanup_jobs_.erase(it);
|
|
} else
|
|
++it;
|
|
for (auto it = sessions_.begin(); it != sessions_.end();) {
|
|
const auto& s = it->second;
|
|
const bool pending =
|
|
std::any_of(auth_jobs_.begin(), auth_jobs_.end(),
|
|
[&](const auto& j) { return j.session == s; }) ||
|
|
std::any_of(prepare_jobs_.begin(), prepare_jobs_.end(),
|
|
[&](const auto& j) { return j.session == s; });
|
|
if (s->closing && !pending) {
|
|
s->phase = Session::Closed;
|
|
it = sessions_.erase(it);
|
|
} else
|
|
++it;
|
|
}
|
|
if (stopping_ && !network_drained_ && sessions_.empty() &&
|
|
auth_jobs_.empty() && prepare_jobs_.empty() && cleanup_jobs_.empty() &&
|
|
!reset_job_.valid()) {
|
|
transport_.Shutdown();
|
|
network_drained_ = true;
|
|
}
|
|
}
|
|
} // namespace crossdesk::lan
|