mirror of
https://github.com/kunkundi/crossdesk.git
synced 2026-10-28 00:33:06 +08:00
[feat] add authenticated LAN transport and per-IP listeners
This commit is contained in:
dijunkun
@@ -0,0 +1,806 @@
|
||||
#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
|
||||
Reference in New Issue
Block a user