node_controller.cc 51 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474
  1. // Copyright 2016 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "mojo/core/node_controller.h"
  5. #include <algorithm>
  6. #include <limits>
  7. #include <vector>
  8. #include "base/bind.h"
  9. #include "base/containers/queue.h"
  10. #include "base/location.h"
  11. #include "base/logging.h"
  12. #include "base/process/process_handle.h"
  13. #include "base/rand_util.h"
  14. #include "base/strings/string_piece.h"
  15. #include "base/task/current_thread.h"
  16. #include "build/build_config.h"
  17. #include "mojo/core/broker.h"
  18. #include "mojo/core/broker_host.h"
  19. #include "mojo/core/configuration.h"
  20. #include "mojo/core/ports/name.h"
  21. #include "mojo/core/ports/port_locker.h"
  22. #include "mojo/core/request_context.h"
  23. #include "mojo/core/user_message_impl.h"
  24. #include "mojo/public/cpp/platform/named_platform_channel.h"
  25. #include "mojo/public/cpp/platform/platform_channel.h"
  26. #if BUILDFLAG(IS_WIN)
  27. #include <windows.h>
  28. #endif
  29. #if !BUILDFLAG(IS_NACL)
  30. #include "crypto/random.h"
  31. #endif
  32. namespace mojo {
  33. namespace core {
  34. namespace {
  35. #if BUILDFLAG(IS_NACL)
  36. template <typename T>
  37. void GenerateRandomName(T* out) {
  38. base::RandBytes(out, sizeof(T));
  39. }
  40. #else
  41. template <typename T>
  42. void GenerateRandomName(T* out) {
  43. crypto::RandBytes(out, sizeof(T));
  44. }
  45. #endif
  46. ports::NodeName GetRandomNodeName() {
  47. ports::NodeName name;
  48. GenerateRandomName(&name);
  49. return name;
  50. }
  51. Channel::MessagePtr SerializeEventMessage(ports::ScopedEvent event) {
  52. if (event->type() == ports::Event::Type::kUserMessage) {
  53. // User message events must already be partially serialized.
  54. return UserMessageImpl::FinalizeEventMessage(
  55. ports::Event::Cast<ports::UserMessageEvent>(&event));
  56. }
  57. void* data;
  58. size_t size = event->GetSerializedSize();
  59. auto message = NodeChannel::CreateEventMessage(size, size, &data, 0);
  60. event->Serialize(data);
  61. return message;
  62. }
  63. ports::ScopedEvent DeserializeEventMessage(
  64. const ports::NodeName& from_node,
  65. Channel::MessagePtr channel_message) {
  66. void* data;
  67. size_t size;
  68. bool valid = NodeChannel::GetEventMessageData(*channel_message, &data, &size);
  69. if (!valid)
  70. return nullptr;
  71. auto event = ports::Event::Deserialize(data, size);
  72. if (!event)
  73. return nullptr;
  74. if (event->type() != ports::Event::Type::kUserMessage)
  75. return event;
  76. // User messages require extra parsing.
  77. const size_t event_size = event->GetSerializedSize();
  78. // Note that if this weren't true, the event couldn't have been deserialized
  79. // in the first place.
  80. DCHECK_LE(event_size, size);
  81. auto message_event = ports::Event::Cast<ports::UserMessageEvent>(&event);
  82. auto message = UserMessageImpl::CreateFromChannelMessage(
  83. message_event.get(), std::move(channel_message),
  84. static_cast<uint8_t*>(data) + event_size, size - event_size);
  85. if (!message)
  86. return nullptr;
  87. message->set_source_node(from_node);
  88. message_event->AttachMessage(std::move(message));
  89. return std::move(message_event);
  90. }
  91. // Used by NodeController to watch for shutdown. Since no IO can happen once
  92. // the IO thread is killed, the NodeController can cleanly drop all its peers
  93. // at that time.
  94. class ThreadDestructionObserver
  95. : public base::CurrentThread::DestructionObserver {
  96. public:
  97. static void Create(scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  98. base::OnceClosure callback) {
  99. if (task_runner->RunsTasksInCurrentSequence()) {
  100. // Owns itself.
  101. new ThreadDestructionObserver(std::move(callback));
  102. } else {
  103. task_runner->PostTask(
  104. FROM_HERE, base::BindOnce(&Create, task_runner, std::move(callback)));
  105. }
  106. }
  107. ThreadDestructionObserver(const ThreadDestructionObserver&) = delete;
  108. ThreadDestructionObserver& operator=(const ThreadDestructionObserver&) =
  109. delete;
  110. private:
  111. explicit ThreadDestructionObserver(base::OnceClosure callback)
  112. : callback_(std::move(callback)) {
  113. base::CurrentThread::Get()->AddDestructionObserver(this);
  114. }
  115. ~ThreadDestructionObserver() override {
  116. base::CurrentThread::Get()->RemoveDestructionObserver(this);
  117. }
  118. // base::CurrentThread::DestructionObserver:
  119. void WillDestroyCurrentMessageLoop() override {
  120. std::move(callback_).Run();
  121. delete this;
  122. }
  123. base::OnceClosure callback_;
  124. };
  125. #if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  126. ConnectionParams CreateSyncNodeConnectionParams(
  127. const base::Process& target_process,
  128. ConnectionParams connection_params,
  129. const ProcessErrorCallback& process_error_callback,
  130. Channel::HandlePolicy& handle_policy) {
  131. ConnectionParams node_connection_params;
  132. const bool is_untrusted_process = connection_params.is_untrusted_process();
  133. // BrokerHost owns itself.
  134. BrokerHost* broker_host = new BrokerHost(
  135. target_process.IsValid() ? target_process.Duplicate() : base::Process(),
  136. std::move(connection_params), process_error_callback);
  137. #if BUILDFLAG(IS_WIN)
  138. // On Windows, if target_process is invalid it means it's elevated or running
  139. // in another session so a named pipe should be used instead.
  140. if (!target_process.IsValid()) {
  141. handle_policy = Channel::HandlePolicy::kRejectHandles;
  142. NamedPlatformChannel::Options options;
  143. NamedPlatformChannel named_channel(options);
  144. node_connection_params =
  145. ConnectionParams(named_channel.TakeServerEndpoint());
  146. node_connection_params.set_is_untrusted_process(is_untrusted_process);
  147. broker_host->SendNamedChannel(named_channel.GetServerName());
  148. return node_connection_params;
  149. }
  150. #endif
  151. // Sync connections usurp the passed endpoint and use it for the sync broker
  152. // channel. A new channel is created here for the NodeChannel and sent over
  153. // a sync broker message to the client.
  154. PlatformChannel node_channel;
  155. node_connection_params = ConnectionParams(node_channel.TakeLocalEndpoint());
  156. node_connection_params.set_is_untrusted_process(is_untrusted_process);
  157. bool channel_ok = broker_host->SendChannel(
  158. node_channel.TakeRemoteEndpoint().TakePlatformHandle());
  159. DCHECK(channel_ok);
  160. return node_connection_params;
  161. }
  162. #endif // !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  163. } // namespace
  164. NodeController::~NodeController() = default;
  165. NodeController::NodeController()
  166. : name_(GetRandomNodeName()), node_(new ports::Node(name_, this)) {
  167. DVLOG(1) << "Initializing node " << name_;
  168. }
  169. void NodeController::SetIOTaskRunner(
  170. scoped_refptr<base::SingleThreadTaskRunner> task_runner) {
  171. io_task_runner_ = task_runner;
  172. ThreadDestructionObserver::Create(
  173. io_task_runner_,
  174. base::BindOnce(&NodeController::DropAllPeers, base::Unretained(this)));
  175. }
  176. void NodeController::SendBrokerClientInvitation(
  177. base::Process target_process,
  178. ConnectionParams connection_params,
  179. const std::vector<std::pair<std::string, ports::PortRef>>& attached_ports,
  180. const ProcessErrorCallback& process_error_callback) {
  181. // Generate the temporary remote node name here so that it can be associated
  182. // with the ports "attached" to this invitation.
  183. ports::NodeName temporary_node_name;
  184. GenerateRandomName(&temporary_node_name);
  185. {
  186. base::AutoLock lock(reserved_ports_lock_);
  187. PortMap& port_map = reserved_ports_[temporary_node_name];
  188. for (auto& entry : attached_ports) {
  189. auto result = port_map.emplace(entry.first, entry.second);
  190. DCHECK(result.second) << "Duplicate attachment: " << entry.first;
  191. }
  192. }
  193. io_task_runner_->PostTask(
  194. FROM_HERE,
  195. base::BindOnce(&NodeController::SendBrokerClientInvitationOnIOThread,
  196. base::Unretained(this), std::move(target_process),
  197. std::move(connection_params), temporary_node_name,
  198. process_error_callback));
  199. }
  200. void NodeController::AcceptBrokerClientInvitation(
  201. ConnectionParams connection_params) {
  202. absl::optional<PlatformHandle> broker_host_handle;
  203. DCHECK(!GetConfiguration().is_broker_process);
  204. #if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  205. if (!connection_params.is_async()) {
  206. // Use the bootstrap channel for the broker and receive the node's channel
  207. // synchronously as the first message from the broker.
  208. DCHECK(connection_params.endpoint().is_valid());
  209. broker_ = std::make_unique<Broker>(
  210. connection_params.TakeEndpoint().TakePlatformHandle(),
  211. /*wait_for_channel_handle=*/true);
  212. PlatformChannelEndpoint endpoint = broker_->GetInviterEndpoint();
  213. if (!endpoint.is_valid()) {
  214. // Most likely the inviter's side of the channel has already been closed
  215. // and the broker was unable to negotiate a NodeChannel pipe. In this case
  216. // we can cancel our connection to our inviter.
  217. DVLOG(1) << "Cannot connect to invalid inviter channel.";
  218. CancelPendingPortMerges();
  219. return;
  220. }
  221. const bool leak_endpoint = connection_params.leak_endpoint();
  222. connection_params = ConnectionParams(std::move(endpoint));
  223. connection_params.set_leak_endpoint(leak_endpoint);
  224. } else {
  225. // For async connections, we instead create a new channel for the broker and
  226. // send a request for the inviting process to bind to it. This avoids doing
  227. // blocking I/O to accept the invitation. Does not work in some sandboxed
  228. // environments, where the PlatformChannel constructor will CHECK fail.
  229. PlatformChannel channel;
  230. broker_ = std::make_unique<Broker>(
  231. channel.TakeLocalEndpoint().TakePlatformHandle(),
  232. /*wait_for_channel_handle=*/false);
  233. broker_host_handle = channel.TakeRemoteEndpoint().TakePlatformHandle();
  234. }
  235. #endif
  236. // Re-enable port merge operations, which may have been disabled if this isn't
  237. // the first invitation accepted by this process.
  238. base::AutoLock lock(pending_port_merges_lock_);
  239. reject_pending_merges_ = false;
  240. io_task_runner_->PostTask(
  241. FROM_HERE,
  242. base::BindOnce(&NodeController::AcceptBrokerClientInvitationOnIOThread,
  243. base::Unretained(this), std::move(connection_params),
  244. std::move(broker_host_handle)));
  245. }
  246. void NodeController::ConnectIsolated(ConnectionParams connection_params,
  247. const ports::PortRef& port,
  248. base::StringPiece connection_name) {
  249. io_task_runner_->PostTask(
  250. FROM_HERE,
  251. base::BindOnce(&NodeController::ConnectIsolatedOnIOThread,
  252. base::Unretained(this), std::move(connection_params), port,
  253. std::string(connection_name)));
  254. }
  255. void NodeController::SetPortObserver(const ports::PortRef& port,
  256. scoped_refptr<PortObserver> observer) {
  257. node_->SetUserData(port, std::move(observer));
  258. }
  259. void NodeController::ClosePort(const ports::PortRef& port) {
  260. SetPortObserver(port, nullptr);
  261. int rv = node_->ClosePort(port);
  262. DCHECK_EQ(rv, ports::OK) << " Failed to close port: " << port.name();
  263. }
  264. int NodeController::SendUserMessage(
  265. const ports::PortRef& port,
  266. std::unique_ptr<ports::UserMessageEvent> message) {
  267. return node_->SendUserMessage(port, std::move(message));
  268. }
  269. void NodeController::MergePortIntoInviter(const std::string& name,
  270. const ports::PortRef& port) {
  271. scoped_refptr<NodeChannel> inviter;
  272. bool reject_merge = false;
  273. {
  274. // Hold |pending_port_merges_lock_| while getting |inviter|. Otherwise,
  275. // there is a race where the inviter can be set, and |pending_port_merges_|
  276. // be processed between retrieving |inviter| and adding the merge to
  277. // |pending_port_merges_|.
  278. base::AutoLock lock(pending_port_merges_lock_);
  279. inviter = GetInviterChannel();
  280. if (reject_pending_merges_) {
  281. reject_merge = true;
  282. } else if (!inviter) {
  283. pending_port_merges_.push_back(std::make_pair(name, port));
  284. return;
  285. }
  286. }
  287. if (reject_merge) {
  288. node_->ClosePort(port);
  289. DVLOG(2) << "Rejecting port merge for name " << name
  290. << " due to closed inviter channel.";
  291. return;
  292. }
  293. RecordPendingPortMerge(port);
  294. inviter->RequestPortMerge(port.name(), name);
  295. }
  296. int NodeController::MergeLocalPorts(const ports::PortRef& port0,
  297. const ports::PortRef& port1) {
  298. return node_->MergeLocalPorts(port0, port1);
  299. }
  300. base::WritableSharedMemoryRegion NodeController::CreateSharedBuffer(
  301. size_t num_bytes) {
  302. #if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  303. // Shared buffer creation failure is fatal, so always use the broker when we
  304. // have one; unless of course the embedder forces us not to.
  305. if (!GetConfiguration().force_direct_shared_memory_allocation && broker_)
  306. return broker_->GetWritableSharedMemoryRegion(num_bytes);
  307. #endif
  308. return base::WritableSharedMemoryRegion::Create(num_bytes);
  309. }
  310. void NodeController::RequestShutdown(base::OnceClosure callback) {
  311. {
  312. base::AutoLock lock(shutdown_lock_);
  313. shutdown_callback_ = std::move(callback);
  314. shutdown_callback_flag_.Set(true);
  315. }
  316. AttemptShutdownIfRequested();
  317. }
  318. void NodeController::NotifyBadMessageFrom(const ports::NodeName& source_node,
  319. const std::string& error) {
  320. scoped_refptr<NodeChannel> peer = GetPeerChannel(source_node);
  321. DCHECK(peer);
  322. DCHECK(peer->HasBadMessageHandler());
  323. peer->NotifyBadMessage(error);
  324. }
  325. bool NodeController::HasBadMessageHandler(const ports::NodeName& source_node) {
  326. scoped_refptr<NodeChannel> peer = GetPeerChannel(source_node);
  327. return peer ? peer->HasBadMessageHandler() : false;
  328. }
  329. void NodeController::ForceDisconnectProcessForTesting(
  330. base::ProcessId process_id) {
  331. io_task_runner_->PostTask(
  332. FROM_HERE,
  333. base::BindOnce(
  334. &NodeController::ForceDisconnectProcessForTestingOnIOThread,
  335. base::Unretained(this), process_id));
  336. }
  337. void NodeController::RecordPendingPortMerge(const ports::PortRef& port_ref) {
  338. // TODO(sroettger): this should also keep track of the node that is allowed
  339. // to trigger the merge.
  340. ports::SinglePortLocker locker(&port_ref);
  341. locker.port()->pending_merge_peer = true;
  342. }
  343. // static
  344. void NodeController::DeserializeRawBytesAsEventForFuzzer(
  345. base::span<const unsigned char> data) {
  346. void* payload;
  347. auto message = NodeChannel::CreateEventMessage(0, data.size(), &payload, 0);
  348. DCHECK(message);
  349. std::copy(data.begin(), data.end(), static_cast<unsigned char*>(payload));
  350. DeserializeEventMessage(ports::NodeName(), std::move(message));
  351. }
  352. // static
  353. void NodeController::DeserializeMessageAsEventForFuzzer(
  354. Channel::MessagePtr message) {
  355. DeserializeEventMessage(ports::NodeName(), std::move(message));
  356. }
  357. void NodeController::SendBrokerClientInvitationOnIOThread(
  358. base::Process target_process,
  359. ConnectionParams connection_params,
  360. ports::NodeName temporary_node_name,
  361. const ProcessErrorCallback& process_error_callback) {
  362. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  363. #if !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  364. Channel::HandlePolicy handle_policy = Channel::HandlePolicy::kAcceptHandles;
  365. ConnectionParams node_connection_params;
  366. if (connection_params.is_async()) {
  367. // For async connections, the passed endpoint really is the NodeChannel
  368. // endpoint. The broker channel will be established asynchronously by a
  369. // |BIND_SYNC_BROKER| message from the invited client.
  370. node_connection_params = std::move(connection_params);
  371. } else {
  372. node_connection_params = CreateSyncNodeConnectionParams(
  373. target_process, std::move(connection_params), process_error_callback,
  374. handle_policy);
  375. }
  376. scoped_refptr<NodeChannel> channel = NodeChannel::Create(
  377. this, std::move(node_connection_params), handle_policy, io_task_runner_,
  378. process_error_callback);
  379. #else // !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  380. scoped_refptr<NodeChannel> channel = NodeChannel::Create(
  381. this, std::move(connection_params), Channel::HandlePolicy::kAcceptHandles,
  382. io_task_runner_, process_error_callback);
  383. #endif // !BUILDFLAG(IS_APPLE) && !BUILDFLAG(IS_NACL) && !BUILDFLAG(IS_FUCHSIA)
  384. // We set up the invitee channel with a temporary name so it can be identified
  385. // as a pending invitee if it writes any messages to the channel. We may start
  386. // receiving messages from it (though we shouldn't) as soon as Start() is
  387. // called below.
  388. pending_invitations_.insert(std::make_pair(temporary_node_name, channel));
  389. channel->SetRemoteNodeName(temporary_node_name);
  390. channel->SetRemoteProcessHandle(std::move(target_process));
  391. channel->Start();
  392. channel->AcceptInvitee(name_, temporary_node_name);
  393. }
  394. void NodeController::AcceptBrokerClientInvitationOnIOThread(
  395. ConnectionParams connection_params,
  396. absl::optional<PlatformHandle> broker_host_handle) {
  397. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  398. {
  399. base::AutoLock lock(inviter_lock_);
  400. if (inviter_name_ != ports::kInvalidNodeName) {
  401. // We've already accepted an invitation before and are already part of
  402. // a different Mojo process network. In order to accept this new one and
  403. // remain in a consistent state, we have to purge all peer connections and
  404. // start from scratch.
  405. {
  406. base::AutoUnlock unlock(inviter_lock_);
  407. DropAllPeers();
  408. }
  409. inviter_name_ = ports::kInvalidNodeName;
  410. }
  411. const bool leak_endpoint = connection_params.leak_endpoint();
  412. // At this point we don't know the inviter's name, so we can't yet insert it
  413. // into our |peers_| map. That will happen as soon as we receive an
  414. // AcceptInvitee message from them.
  415. bootstrap_inviter_channel_ =
  416. NodeChannel::Create(this, std::move(connection_params),
  417. Channel::HandlePolicy::kAcceptHandles,
  418. io_task_runner_, ProcessErrorCallback());
  419. if (leak_endpoint) {
  420. // Prevent the inviter pipe handle from being closed on shutdown. Pipe
  421. // closure may be used by the inviter to detect that the invited process
  422. // has terminated. In such cases, the invited process must not be invited
  423. // more than once in its lifetime; otherwise this leak matters.
  424. //
  425. // Note that this behavior is supported primarily to help adapt legacy
  426. // Chrome IPC to Mojo, since channel disconnection is used there as a
  427. // signal for normal child process termination.
  428. bootstrap_inviter_channel_->LeakHandleOnShutdown();
  429. }
  430. }
  431. bootstrap_inviter_channel_->Start();
  432. if (broker_host_handle)
  433. bootstrap_inviter_channel_->BindBrokerHost(std::move(*broker_host_handle));
  434. }
  435. void NodeController::ConnectIsolatedOnIOThread(
  436. ConnectionParams connection_params,
  437. ports::PortRef port,
  438. const std::string& connection_name) {
  439. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  440. // Processes using isolated connections to communicate have no ability to lean
  441. // on a broker for handle relaying, so we allow them to send handles to each
  442. // other at their own peril.
  443. scoped_refptr<NodeChannel> channel = NodeChannel::Create(
  444. this, std::move(connection_params), Channel::HandlePolicy::kAcceptHandles,
  445. io_task_runner_, {});
  446. RequestContext request_context;
  447. ports::NodeName token;
  448. GenerateRandomName(&token);
  449. pending_isolated_connections_.emplace(
  450. token, IsolatedConnection{channel, port, connection_name});
  451. if (!connection_name.empty()) {
  452. // If a connection already exists with this name, drop it.
  453. auto it = named_isolated_connections_.find(connection_name);
  454. if (it != named_isolated_connections_.end()) {
  455. ports::NodeName connection_node = it->second;
  456. if (connection_node != name_) {
  457. DropPeer(connection_node, nullptr);
  458. } else {
  459. auto pending_it = pending_isolated_connections_.find(connection_node);
  460. if (pending_it != pending_isolated_connections_.end()) {
  461. node_->ClosePort(pending_it->second.local_port);
  462. pending_isolated_connections_.erase(pending_it);
  463. }
  464. named_isolated_connections_.erase(it);
  465. }
  466. }
  467. named_isolated_connections_.emplace(connection_name, token);
  468. }
  469. channel->SetRemoteNodeName(token);
  470. channel->Start();
  471. RecordPendingPortMerge(port);
  472. channel->AcceptPeer(name_, token, port.name());
  473. }
  474. scoped_refptr<NodeChannel> NodeController::GetPeerChannel(
  475. const ports::NodeName& name) {
  476. base::AutoLock lock(peers_lock_);
  477. auto it = peers_.find(name);
  478. if (it == peers_.end())
  479. return nullptr;
  480. return it->second;
  481. }
  482. scoped_refptr<NodeChannel> NodeController::GetInviterChannel() {
  483. ports::NodeName inviter_name;
  484. {
  485. base::AutoLock lock(inviter_lock_);
  486. inviter_name = inviter_name_;
  487. }
  488. return GetPeerChannel(inviter_name);
  489. }
  490. scoped_refptr<NodeChannel> NodeController::GetBrokerChannel() {
  491. if (GetConfiguration().is_broker_process)
  492. return nullptr;
  493. ports::NodeName broker_name;
  494. {
  495. base::AutoLock lock(broker_lock_);
  496. broker_name = broker_name_;
  497. }
  498. return GetPeerChannel(broker_name);
  499. }
  500. void NodeController::AddPeer(const ports::NodeName& name,
  501. scoped_refptr<NodeChannel> channel,
  502. bool start_channel,
  503. bool allow_name_reuse) {
  504. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  505. DCHECK(name != ports::kInvalidNodeName);
  506. DCHECK(channel);
  507. channel->SetRemoteNodeName(name);
  508. OutgoingMessageQueue pending_messages;
  509. {
  510. base::AutoLock lock(peers_lock_);
  511. if (peers_.find(name) != peers_.end()) {
  512. // This can happen normally if two nodes race to be introduced to each
  513. // other. The losing pipe will be silently closed and introduction should
  514. // not be affected.
  515. DVLOG(1) << "Ignoring duplicate peer name " << name;
  516. return;
  517. }
  518. if (dropped_peers_.Contains(name) && !allow_name_reuse) {
  519. DVLOG(1) << "Trying to re-add dropped peer " << name;
  520. return;
  521. }
  522. auto result = peers_.insert(std::make_pair(name, channel));
  523. DCHECK(result.second);
  524. DVLOG(2) << "Accepting new peer " << name << " on node " << name_;
  525. auto it = pending_peer_messages_.find(name);
  526. if (it != pending_peer_messages_.end()) {
  527. std::swap(pending_messages, it->second);
  528. pending_peer_messages_.erase(it);
  529. }
  530. }
  531. if (start_channel)
  532. channel->Start();
  533. // Flush any queued message we need to deliver to this node.
  534. while (!pending_messages.empty()) {
  535. channel->SendChannelMessage(std::move(pending_messages.front()));
  536. pending_messages.pop();
  537. }
  538. }
  539. void NodeController::DropPeer(const ports::NodeName& node_name,
  540. NodeChannel* channel) {
  541. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  542. // NOTE: Either the `peers_` erasure or the `pending_invitations_` erasure
  543. // below, if executed, may drop the last reference to the named NodeChannel
  544. // and thus result in its deletion. The passed `node_name` argument may be
  545. // owned by that same NodeChannel, so we make a copy of it here to avoid
  546. // potentially unsafe references further below.
  547. ports::NodeName name = node_name;
  548. {
  549. base::AutoLock lock(peers_lock_);
  550. auto it = peers_.find(name);
  551. if (it != peers_.end()) {
  552. ports::NodeName peer = it->first;
  553. peers_.erase(it);
  554. dropped_peers_.Insert(peer);
  555. DVLOG(1) << "Dropped peer " << peer;
  556. }
  557. pending_peer_messages_.erase(name);
  558. pending_invitations_.erase(name);
  559. }
  560. std::vector<ports::PortRef> ports_to_close;
  561. {
  562. // Clean up any reserved ports.
  563. base::AutoLock lock(reserved_ports_lock_);
  564. auto it = reserved_ports_.find(name);
  565. if (it != reserved_ports_.end()) {
  566. for (auto& entry : it->second)
  567. ports_to_close.emplace_back(entry.second);
  568. reserved_ports_.erase(it);
  569. }
  570. }
  571. bool is_inviter;
  572. {
  573. base::AutoLock lock(inviter_lock_);
  574. is_inviter = (name == inviter_name_ ||
  575. (channel && channel == bootstrap_inviter_channel_));
  576. }
  577. // If the error comes from the inviter channel, we also need to cancel any
  578. // port merge requests, so that errors can be propagated to the message
  579. // pipes.
  580. if (is_inviter)
  581. CancelPendingPortMerges();
  582. auto connection_it = pending_isolated_connections_.find(name);
  583. if (connection_it != pending_isolated_connections_.end()) {
  584. IsolatedConnection& connection = connection_it->second;
  585. ports_to_close.push_back(connection.local_port);
  586. if (!connection.name.empty())
  587. named_isolated_connections_.erase(connection.name);
  588. pending_isolated_connections_.erase(connection_it);
  589. }
  590. for (const auto& port : ports_to_close)
  591. node_->ClosePort(port);
  592. node_->LostConnectionToNode(name);
  593. AttemptShutdownIfRequested();
  594. }
  595. void NodeController::SendPeerEvent(const ports::NodeName& name,
  596. ports::ScopedEvent event) {
  597. Channel::MessagePtr event_message = SerializeEventMessage(std::move(event));
  598. if (!event_message)
  599. return;
  600. scoped_refptr<NodeChannel> peer = GetPeerChannel(name);
  601. #if BUILDFLAG(IS_WIN)
  602. if (event_message->has_handles()) {
  603. // If we're sending a message with handles we aren't the destination
  604. // node's inviter or broker (i.e. we don't know its process handle), ask
  605. // the broker to relay for us.
  606. scoped_refptr<NodeChannel> broker = GetBrokerChannel();
  607. if (!peer || !peer->HasRemoteProcessHandle()) {
  608. if (!GetConfiguration().is_broker_process && broker) {
  609. broker->RelayEventMessage(name, std::move(event_message));
  610. } else {
  611. base::AutoLock lock(broker_lock_);
  612. pending_relay_messages_[name].emplace(std::move(event_message));
  613. }
  614. return;
  615. }
  616. }
  617. #endif // BUILDFLAG(IS_WIN)
  618. if (peer) {
  619. peer->SendChannelMessage(std::move(event_message));
  620. return;
  621. }
  622. // If we don't know who the peer is and we are the broker, we can only assume
  623. // the peer is invalid, i.e., it's either a junk name or has already been
  624. // disconnected.
  625. scoped_refptr<NodeChannel> broker = GetBrokerChannel();
  626. if (!broker) {
  627. DVLOG(1) << "Dropping message for unknown peer: " << name;
  628. return;
  629. }
  630. // If we aren't the broker, assume we just need to be introduced and queue
  631. // until that can be either confirmed or denied by the broker.
  632. bool needs_introduction = false;
  633. {
  634. base::AutoLock lock(peers_lock_);
  635. // We may have been introduced on another thread by the time we get here.
  636. // Double-check to be safe.
  637. auto it = peers_.find(name);
  638. if (it == peers_.end()) {
  639. auto& queue = pending_peer_messages_[name];
  640. needs_introduction = queue.empty();
  641. queue.emplace(std::move(event_message));
  642. } else {
  643. peer = it->second;
  644. }
  645. }
  646. if (needs_introduction)
  647. broker->RequestIntroduction(name);
  648. else if (peer)
  649. peer->SendChannelMessage(std::move(event_message));
  650. }
  651. void NodeController::DropAllPeers() {
  652. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  653. std::vector<scoped_refptr<NodeChannel>> all_peers;
  654. {
  655. base::AutoLock lock(inviter_lock_);
  656. if (bootstrap_inviter_channel_) {
  657. // |bootstrap_inviter_channel_| isn't null'd here because we rely on its
  658. // existence to determine whether or not this is the root node. Once
  659. // bootstrap_inviter_channel_->ShutDown() has been called,
  660. // |bootstrap_inviter_channel_| is essentially a dead object and it
  661. // doesn't matter if it's deleted now or when |this| is deleted. Note:
  662. // |bootstrap_inviter_channel_| is only modified on the IO thread.
  663. all_peers.push_back(bootstrap_inviter_channel_);
  664. }
  665. }
  666. {
  667. base::AutoLock lock(peers_lock_);
  668. for (const auto& peer : peers_)
  669. all_peers.push_back(peer.second);
  670. for (const auto& peer : pending_invitations_)
  671. all_peers.push_back(peer.second);
  672. peers_.clear();
  673. pending_invitations_.clear();
  674. pending_peer_messages_.clear();
  675. pending_isolated_connections_.clear();
  676. named_isolated_connections_.clear();
  677. }
  678. for (const auto& peer : all_peers)
  679. peer->ShutDown();
  680. if (destroy_on_io_thread_shutdown_)
  681. delete this;
  682. }
  683. void NodeController::ForwardEvent(const ports::NodeName& node,
  684. ports::ScopedEvent event) {
  685. DCHECK(event);
  686. if (node == name_)
  687. node_->AcceptEvent(name_, std::move(event));
  688. else
  689. SendPeerEvent(node, std::move(event));
  690. AttemptShutdownIfRequested();
  691. }
  692. void NodeController::BroadcastEvent(ports::ScopedEvent event) {
  693. Channel::MessagePtr channel_message = SerializeEventMessage(std::move(event));
  694. DCHECK(channel_message && !channel_message->has_handles());
  695. scoped_refptr<NodeChannel> broker = GetBrokerChannel();
  696. if (broker) {
  697. broker->Broadcast(std::move(channel_message));
  698. } else if (broker_name_ == ports::kInvalidNodeName) {
  699. // Do an additional check if broker_name_ is not set. It's possible that we
  700. // don't have a broker channel even though we're not the broker ourselves,
  701. // e.g. if this code path is called from the channel error path..
  702. OnBroadcast(name_, std::move(channel_message));
  703. }
  704. }
  705. void NodeController::PortStatusChanged(const ports::PortRef& port) {
  706. scoped_refptr<ports::UserData> user_data;
  707. node_->GetUserData(port, &user_data);
  708. PortObserver* observer = static_cast<PortObserver*>(user_data.get());
  709. if (observer) {
  710. observer->OnPortStatusChanged();
  711. } else {
  712. DVLOG(2) << "Ignoring status change for " << port.name() << " because it "
  713. << "doesn't have an observer.";
  714. }
  715. }
  716. void NodeController::OnAcceptInvitee(const ports::NodeName& from_node,
  717. const ports::NodeName& inviter_name,
  718. const ports::NodeName& token) {
  719. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  720. scoped_refptr<NodeChannel> inviter;
  721. {
  722. base::AutoLock lock(inviter_lock_);
  723. if (bootstrap_inviter_channel_ &&
  724. inviter_name_ == ports::kInvalidNodeName) {
  725. inviter_name_ = inviter_name;
  726. inviter = bootstrap_inviter_channel_;
  727. }
  728. }
  729. if (!inviter) {
  730. DLOG(ERROR) << "Unexpected AcceptInvitee message from " << from_node;
  731. DropPeer(from_node, nullptr);
  732. return;
  733. }
  734. inviter->SetRemoteNodeName(inviter_name);
  735. inviter->AcceptInvitation(token, name_);
  736. // NOTE: The invitee does not actually add its inviter as a peer until
  737. // receiving an AcceptBrokerClient message from the broker. The inviter will
  738. // request that said message be sent upon receiving AcceptInvitation.
  739. DVLOG(1) << "Broker client " << name_ << " accepting invitation from "
  740. << inviter_name;
  741. }
  742. void NodeController::OnAcceptInvitation(const ports::NodeName& from_node,
  743. const ports::NodeName& token,
  744. const ports::NodeName& invitee_name) {
  745. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  746. auto it = pending_invitations_.find(from_node);
  747. if (it == pending_invitations_.end() || token != from_node) {
  748. DLOG(ERROR) << "Received unexpected AcceptInvitation message from "
  749. << from_node;
  750. DropPeer(from_node, nullptr);
  751. return;
  752. }
  753. {
  754. base::AutoLock lock(reserved_ports_lock_);
  755. auto reserved_ports_it = reserved_ports_.find(from_node);
  756. if (reserved_ports_it != reserved_ports_.end()) {
  757. // Swap the temporary node name's reserved ports into an entry keyed by
  758. // the real node name.
  759. auto result = reserved_ports_.emplace(
  760. invitee_name, std::move(reserved_ports_it->second));
  761. DCHECK(result.second);
  762. reserved_ports_.erase(reserved_ports_it);
  763. }
  764. }
  765. scoped_refptr<NodeChannel> channel = it->second;
  766. pending_invitations_.erase(it);
  767. DCHECK(channel);
  768. DVLOG(1) << "Node " << name_ << " accepted invitee " << invitee_name;
  769. AddPeer(invitee_name, channel, false /* start_channel */);
  770. // TODO(rockot): We could simplify invitee initialization if we could
  771. // synchronously get a new async broker channel from the broker. For now we do
  772. // it asynchronously since it's only used to facilitate handle passing, not
  773. // handle creation.
  774. scoped_refptr<NodeChannel> broker = GetBrokerChannel();
  775. if (broker) {
  776. // Inform the broker of this new client.
  777. broker->AddBrokerClient(invitee_name, channel->CloneRemoteProcessHandle());
  778. } else {
  779. // If we have no broker, either we need to wait for one, or we *are* the
  780. // broker.
  781. scoped_refptr<NodeChannel> inviter = GetInviterChannel();
  782. if (!inviter) {
  783. base::AutoLock lock(inviter_lock_);
  784. inviter = bootstrap_inviter_channel_;
  785. }
  786. if (!inviter) {
  787. // Yes, we're the broker. We can initialize the client directly.
  788. channel->AcceptBrokerClient(name_, PlatformHandle(),
  789. channel->LocalCapabilities());
  790. } else {
  791. // We aren't the broker, so wait for a broker connection.
  792. base::AutoLock lock(broker_lock_);
  793. pending_broker_clients_.push(invitee_name);
  794. }
  795. }
  796. }
  797. void NodeController::OnAddBrokerClient(const ports::NodeName& from_node,
  798. const ports::NodeName& client_name,
  799. base::ProcessHandle process_handle) {
  800. base::Process scoped_process_handle(process_handle);
  801. scoped_refptr<NodeChannel> sender = GetPeerChannel(from_node);
  802. if (!sender) {
  803. DLOG(ERROR) << "Ignoring AddBrokerClient from unknown sender.";
  804. return;
  805. }
  806. if (!GetConfiguration().is_broker_process) {
  807. DLOG(ERROR) << "Ignoring AddBrokerClient on non-broker node.";
  808. return;
  809. }
  810. if (GetPeerChannel(client_name)) {
  811. DLOG(ERROR) << "Ignoring AddBrokerClient for known client.";
  812. DropPeer(from_node, nullptr);
  813. return;
  814. }
  815. PlatformChannel broker_channel;
  816. ConnectionParams connection_params(broker_channel.TakeLocalEndpoint());
  817. scoped_refptr<NodeChannel> client = NodeChannel::Create(
  818. this, std::move(connection_params), Channel::HandlePolicy::kAcceptHandles,
  819. io_task_runner_, ProcessErrorCallback());
  820. #if BUILDFLAG(IS_WIN)
  821. // The broker must have a working handle to the client process in order to
  822. // properly copy other handles to and from the client.
  823. if (!scoped_process_handle.IsValid()) {
  824. DLOG(ERROR) << "Broker rejecting client with invalid process handle.";
  825. return;
  826. }
  827. #endif
  828. client->SetRemoteProcessHandle(std::move(scoped_process_handle));
  829. AddPeer(client_name, client, true /* start_channel */);
  830. DVLOG(1) << "Broker " << name_ << " accepting client " << client_name
  831. << " from peer " << from_node;
  832. sender->BrokerClientAdded(
  833. client_name, broker_channel.TakeRemoteEndpoint().TakePlatformHandle());
  834. }
  835. void NodeController::OnBrokerClientAdded(const ports::NodeName& from_node,
  836. const ports::NodeName& client_name,
  837. PlatformHandle broker_channel) {
  838. scoped_refptr<NodeChannel> client = GetPeerChannel(client_name);
  839. if (!client) {
  840. DLOG(ERROR) << "BrokerClientAdded for unknown client " << client_name;
  841. return;
  842. }
  843. // This should have come from our own broker.
  844. if (GetBrokerChannel() != GetPeerChannel(from_node)) {
  845. DLOG(ERROR) << "BrokerClientAdded from non-broker node " << from_node;
  846. return;
  847. }
  848. DVLOG(1) << "Client " << client_name << " accepted by broker " << from_node;
  849. client->AcceptBrokerClient(from_node, std::move(broker_channel),
  850. GetBrokerChannel()->RemoteCapabilities());
  851. }
  852. void NodeController::OnAcceptBrokerClient(const ports::NodeName& from_node,
  853. const ports::NodeName& broker_name,
  854. PlatformHandle broker_channel,
  855. const uint64_t broker_capabilities) {
  856. if (GetConfiguration().is_broker_process) {
  857. // The broker should never receive this message from anyone.
  858. DropPeer(from_node, nullptr);
  859. return;
  860. }
  861. // This node should already have an inviter in bootstrap mode.
  862. ports::NodeName inviter_name;
  863. scoped_refptr<NodeChannel> inviter;
  864. {
  865. base::AutoLock lock(inviter_lock_);
  866. inviter_name = inviter_name_;
  867. inviter = bootstrap_inviter_channel_;
  868. bootstrap_inviter_channel_ = nullptr;
  869. }
  870. if (inviter_name != from_node || !inviter ||
  871. broker_name == ports::kInvalidNodeName) {
  872. // We are not expecting this message. Assume the source is hostile.
  873. DropPeer(from_node, nullptr);
  874. return;
  875. }
  876. base::queue<ports::NodeName> pending_broker_clients;
  877. std::unordered_map<ports::NodeName, OutgoingMessageQueue>
  878. pending_relay_messages;
  879. {
  880. base::AutoLock lock(broker_lock_);
  881. broker_name_ = broker_name;
  882. std::swap(pending_broker_clients, pending_broker_clients_);
  883. std::swap(pending_relay_messages, pending_relay_messages_);
  884. }
  885. // It's now possible to add both the broker and the inviter as peers.
  886. // Note that the broker and inviter may be the same node.
  887. scoped_refptr<NodeChannel> broker;
  888. if (broker_name == inviter_name) {
  889. broker = inviter;
  890. } else if (broker_channel.is_valid()) {
  891. broker = NodeChannel::Create(
  892. this,
  893. ConnectionParams(PlatformChannelEndpoint(std::move(broker_channel))),
  894. Channel::HandlePolicy::kAcceptHandles, io_task_runner_,
  895. ProcessErrorCallback());
  896. broker->SetRemoteCapabilities(broker_capabilities);
  897. AddPeer(broker_name, broker, true /* start_channel */);
  898. } else {
  899. DropPeer(from_node, nullptr);
  900. return;
  901. }
  902. AddPeer(inviter_name, inviter, false /* start_channel */);
  903. {
  904. // Complete any port merge requests we have waiting for the inviter.
  905. std::vector<std::pair<std::string, ports::PortRef>> pending_port_merges;
  906. {
  907. base::AutoLock lock(pending_port_merges_lock_);
  908. std::swap(pending_port_merges_, pending_port_merges);
  909. }
  910. std::vector<ports::PortName> pending_port_names;
  911. for (auto& request : pending_port_merges) {
  912. RecordPendingPortMerge(request.second);
  913. inviter->RequestPortMerge(request.second.name(), request.first);
  914. }
  915. }
  916. // Feed the broker any pending invitees of our own.
  917. while (!pending_broker_clients.empty()) {
  918. const ports::NodeName& invitee_name = pending_broker_clients.front();
  919. auto it = peers_.find(invitee_name);
  920. if (it != peers_.end()) {
  921. broker->AddBrokerClient(invitee_name,
  922. it->second->CloneRemoteProcessHandle());
  923. }
  924. pending_broker_clients.pop();
  925. }
  926. #if BUILDFLAG(IS_WIN)
  927. // Have the broker relay any messages we have waiting.
  928. for (auto& entry : pending_relay_messages) {
  929. const ports::NodeName& destination = entry.first;
  930. auto& message_queue = entry.second;
  931. while (!message_queue.empty()) {
  932. broker->RelayEventMessage(destination, std::move(message_queue.front()));
  933. message_queue.pop();
  934. }
  935. }
  936. #endif
  937. if (inviter->HasLocalCapability(kNodeCapabilitySupportsUpgrade) &&
  938. inviter->HasRemoteCapability(kNodeCapabilitySupportsUpgrade)) {
  939. inviter->OfferChannelUpgrade();
  940. }
  941. DVLOG(1) << "Client " << name_ << " accepted by broker " << broker_name;
  942. }
  943. void NodeController::OnEventMessage(const ports::NodeName& from_node,
  944. Channel::MessagePtr channel_message) {
  945. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  946. auto event = DeserializeEventMessage(from_node, std::move(channel_message));
  947. if (!event) {
  948. // We silently ignore unparseable events, as they may come from a process
  949. // running a newer version of Mojo.
  950. DVLOG(1) << "Ignoring invalid or unknown event from " << from_node;
  951. return;
  952. }
  953. node_->AcceptEvent(from_node, std::move(event));
  954. AttemptShutdownIfRequested();
  955. }
  956. void NodeController::OnRequestPortMerge(
  957. const ports::NodeName& from_node,
  958. const ports::PortName& connector_port_name,
  959. const std::string& name) {
  960. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  961. DVLOG(2) << "Node " << name_ << " received RequestPortMerge for name " << name
  962. << " and port " << connector_port_name << "@" << from_node;
  963. ports::PortRef local_port;
  964. {
  965. base::AutoLock lock(reserved_ports_lock_);
  966. auto it = reserved_ports_.find(from_node);
  967. // TODO(https://crbug.com/822034): We should send a notification back to the
  968. // requestor so they can clean up their dangling port in this failure case.
  969. // This requires changes to the internal protocol, which can't be made yet.
  970. // Until this is done, pipes from |MojoExtractMessagePipeFromInvitation()|
  971. // will never break if the given name was invalid.
  972. if (it == reserved_ports_.end()) {
  973. DVLOG(1) << "Ignoring port merge request from node " << from_node << ". "
  974. << "No ports reserved for that node.";
  975. return;
  976. }
  977. PortMap& port_map = it->second;
  978. auto port_it = port_map.find(name);
  979. if (port_it == port_map.end()) {
  980. DVLOG(1) << "Ignoring request to connect to port for unknown name "
  981. << name << " from node " << from_node;
  982. return;
  983. }
  984. local_port = port_it->second;
  985. port_map.erase(port_it);
  986. if (port_map.empty())
  987. reserved_ports_.erase(it);
  988. }
  989. int rv = node_->MergePorts(local_port, from_node, connector_port_name);
  990. if (rv != ports::OK)
  991. DLOG(ERROR) << "MergePorts failed: " << rv;
  992. }
  993. void NodeController::OnRequestIntroduction(const ports::NodeName& from_node,
  994. const ports::NodeName& name) {
  995. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  996. if (broker_name_ != ports::kInvalidNodeName) {
  997. DLOG(ERROR) << "Ignoring OnRequestIntroduction on non-broker node.";
  998. DropPeer(from_node, nullptr);
  999. return;
  1000. }
  1001. scoped_refptr<NodeChannel> requestor = GetPeerChannel(from_node);
  1002. if (from_node == name || name == ports::kInvalidNodeName || !requestor) {
  1003. DLOG(ERROR) << "Rejecting invalid OnRequestIntroduction message from "
  1004. << from_node;
  1005. DropPeer(from_node, nullptr);
  1006. return;
  1007. }
  1008. scoped_refptr<NodeChannel> new_friend = GetPeerChannel(name);
  1009. if (!new_friend) {
  1010. // We don't know who they're talking about!
  1011. requestor->Introduce(name, PlatformHandle(), kNodeCapabilityNone);
  1012. } else {
  1013. PlatformChannel new_channel;
  1014. requestor->Introduce(name,
  1015. new_channel.TakeLocalEndpoint().TakePlatformHandle(),
  1016. new_friend->RemoteCapabilities());
  1017. new_friend->Introduce(from_node,
  1018. new_channel.TakeRemoteEndpoint().TakePlatformHandle(),
  1019. requestor->RemoteCapabilities());
  1020. }
  1021. }
  1022. void NodeController::OnIntroduce(const ports::NodeName& from_node,
  1023. const ports::NodeName& name,
  1024. PlatformHandle channel_handle,
  1025. const uint64_t remote_capabilities) {
  1026. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  1027. if (broker_name_ == ports::kInvalidNodeName || from_node != broker_name_) {
  1028. DVLOG(1) << "Ignoring introduction from non-broker process.";
  1029. DropPeer(from_node, nullptr);
  1030. return;
  1031. }
  1032. if (!channel_handle.is_valid()) {
  1033. node_->LostConnectionToNode(name);
  1034. DVLOG(1) << "Could not be introduced to peer " << name;
  1035. base::AutoLock lock(peers_lock_);
  1036. pending_peer_messages_.erase(name);
  1037. return;
  1038. }
  1039. #if BUILDFLAG(IS_WIN)
  1040. // Introduced peers are never our broker nor our inviter, so we never accept
  1041. // handles from them directly.
  1042. constexpr auto kPeerHandlePolicy = Channel::HandlePolicy::kRejectHandles;
  1043. #else
  1044. constexpr auto kPeerHandlePolicy = Channel::HandlePolicy::kAcceptHandles;
  1045. #endif
  1046. scoped_refptr<NodeChannel> channel = NodeChannel::Create(
  1047. this,
  1048. ConnectionParams(PlatformChannelEndpoint(std::move(channel_handle))),
  1049. kPeerHandlePolicy, io_task_runner_, ProcessErrorCallback());
  1050. DVLOG(1) << "Adding new peer " << name << " via broker introduction.";
  1051. AddPeer(name, channel, true /* start_channel */);
  1052. channel->SetRemoteCapabilities(remote_capabilities);
  1053. if (channel->HasLocalCapability(kNodeCapabilitySupportsUpgrade) &&
  1054. channel->HasRemoteCapability(kNodeCapabilitySupportsUpgrade)) {
  1055. channel->OfferChannelUpgrade();
  1056. }
  1057. }
  1058. void NodeController::OnBroadcast(const ports::NodeName& from_node,
  1059. Channel::MessagePtr message) {
  1060. DCHECK(!message->has_handles());
  1061. if (broker_name_ != ports::kInvalidNodeName) {
  1062. DLOG(ERROR) << "Ignoring OnBroadcast on non-broker node.";
  1063. DropPeer(from_node, nullptr);
  1064. return;
  1065. }
  1066. auto event = DeserializeEventMessage(from_node, std::move(message));
  1067. if (!event) {
  1068. // We silently ignore unparseable events, as they may come from a process
  1069. // running a newer version of Mojo.
  1070. DVLOG(1) << "Ignoring request to broadcast invalid or unknown event from "
  1071. << from_node;
  1072. return;
  1073. }
  1074. base::AutoLock lock(peers_lock_);
  1075. for (auto& iter : peers_) {
  1076. // Clone and send the event to each known peer. Events which cannot be
  1077. // cloned cannot be broadcast.
  1078. ports::ScopedEvent clone = event->CloneForBroadcast();
  1079. if (!clone) {
  1080. DVLOG(1) << "Ignoring request to broadcast invalid event from "
  1081. << from_node << " [type=" << static_cast<uint32_t>(event->type())
  1082. << "]";
  1083. return;
  1084. }
  1085. iter.second->SendChannelMessage(SerializeEventMessage(std::move(clone)));
  1086. }
  1087. }
  1088. #if BUILDFLAG(IS_WIN)
  1089. void NodeController::OnRelayEventMessage(const ports::NodeName& from_node,
  1090. base::ProcessHandle from_process,
  1091. const ports::NodeName& destination,
  1092. Channel::MessagePtr message) {
  1093. // The broker should always know which process this came from.
  1094. DCHECK(from_process != base::kNullProcessHandle);
  1095. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  1096. if (GetBrokerChannel()) {
  1097. // Only the broker should be asked to relay a message.
  1098. LOG(ERROR) << "Non-broker refusing to relay message.";
  1099. DropPeer(from_node, nullptr);
  1100. return;
  1101. }
  1102. if (destination == name_) {
  1103. // Great, we can deliver this message locally.
  1104. OnEventMessage(from_node, std::move(message));
  1105. return;
  1106. }
  1107. scoped_refptr<NodeChannel> peer = GetPeerChannel(destination);
  1108. if (peer)
  1109. peer->EventMessageFromRelay(from_node, std::move(message));
  1110. else
  1111. DLOG(ERROR) << "Dropping relay message for unknown node " << destination;
  1112. }
  1113. void NodeController::OnEventMessageFromRelay(const ports::NodeName& from_node,
  1114. const ports::NodeName& source_node,
  1115. Channel::MessagePtr message) {
  1116. if (GetPeerChannel(from_node) != GetBrokerChannel()) {
  1117. LOG(ERROR) << "Refusing relayed message from non-broker node.";
  1118. DropPeer(from_node, nullptr);
  1119. return;
  1120. }
  1121. OnEventMessage(source_node, std::move(message));
  1122. }
  1123. #endif
  1124. void NodeController::OnAcceptPeer(const ports::NodeName& from_node,
  1125. const ports::NodeName& token,
  1126. const ports::NodeName& peer_name,
  1127. const ports::PortName& port_name) {
  1128. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  1129. auto it = pending_isolated_connections_.find(from_node);
  1130. if (it == pending_isolated_connections_.end()) {
  1131. DLOG(ERROR) << "Received unexpected AcceptPeer message from " << from_node;
  1132. DropPeer(from_node, nullptr);
  1133. return;
  1134. }
  1135. IsolatedConnection& connection = it->second;
  1136. scoped_refptr<NodeChannel> channel = std::move(connection.channel);
  1137. ports::PortRef local_port = connection.local_port;
  1138. if (!connection.name.empty())
  1139. named_isolated_connections_[connection.name] = peer_name;
  1140. pending_isolated_connections_.erase(it);
  1141. DCHECK(channel);
  1142. if (name_ != peer_name) {
  1143. // It's possible (e.g. in tests) that we may "connect" to ourself, in which
  1144. // case we skip this |AddPeer()| call and go straight to merging ports.
  1145. // Note that we explicitly drop any prior connection to the same peer so
  1146. // that new isolated connections can replace old ones.
  1147. DropPeer(peer_name, nullptr);
  1148. AddPeer(peer_name, channel, false /* start_channel */,
  1149. true /* allow_name_reuse */);
  1150. DVLOG(1) << "Node " << name_ << " accepted peer " << peer_name;
  1151. }
  1152. // We need to choose one side to initiate the port merge. It doesn't matter
  1153. // who does it as long as they don't both try. Simple solution: pick the one
  1154. // with the "smaller" port name.
  1155. if (local_port.name() < port_name)
  1156. node()->MergePorts(local_port, peer_name, port_name);
  1157. }
  1158. void NodeController::OnChannelError(const ports::NodeName& from_node,
  1159. NodeChannel* channel) {
  1160. if (io_task_runner_->RunsTasksInCurrentSequence()) {
  1161. RequestContext request_context(RequestContext::Source::SYSTEM);
  1162. DropPeer(from_node, channel);
  1163. } else {
  1164. io_task_runner_->PostTask(
  1165. FROM_HERE,
  1166. base::BindOnce(&NodeController::OnChannelError, base::Unretained(this),
  1167. from_node, base::RetainedRef(channel)));
  1168. }
  1169. }
  1170. void NodeController::CancelPendingPortMerges() {
  1171. std::vector<ports::PortRef> ports_to_close;
  1172. {
  1173. base::AutoLock lock(pending_port_merges_lock_);
  1174. reject_pending_merges_ = true;
  1175. for (const auto& port : pending_port_merges_)
  1176. ports_to_close.push_back(port.second);
  1177. pending_port_merges_.clear();
  1178. }
  1179. for (const auto& port : ports_to_close)
  1180. node_->ClosePort(port);
  1181. }
  1182. void NodeController::DestroyOnIOThreadShutdown() {
  1183. destroy_on_io_thread_shutdown_ = true;
  1184. }
  1185. void NodeController::AttemptShutdownIfRequested() {
  1186. if (!shutdown_callback_flag_)
  1187. return;
  1188. base::OnceClosure callback;
  1189. {
  1190. base::AutoLock lock(shutdown_lock_);
  1191. if (shutdown_callback_.is_null())
  1192. return;
  1193. if (!node_->CanShutdownCleanly(
  1194. ports::Node::ShutdownPolicy::ALLOW_LOCAL_PORTS)) {
  1195. DVLOG(2) << "Unable to cleanly shut down node " << name_;
  1196. return;
  1197. }
  1198. callback = std::move(shutdown_callback_);
  1199. shutdown_callback_flag_.Set(false);
  1200. }
  1201. DCHECK(!callback.is_null());
  1202. std::move(callback).Run();
  1203. }
  1204. void NodeController::ForceDisconnectProcessForTestingOnIOThread(
  1205. base::ProcessId process_id) {
  1206. #if BUILDFLAG(IS_NACL) || BUILDFLAG(IS_IOS)
  1207. NOTREACHED();
  1208. #else
  1209. DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
  1210. RequestContext request_context;
  1211. // A channel may have multiple aliases since we generate one for any we
  1212. // invite and then only later refer to it by its own chosen name.
  1213. NodeMap peers_to_drop;
  1214. for (auto& peer : peers_) {
  1215. NodeChannel* channel = peer.second.get();
  1216. if (channel->HasRemoteProcessHandle()) {
  1217. base::Process process(channel->CloneRemoteProcessHandle());
  1218. if (process.Pid() == process_id)
  1219. peers_to_drop.emplace(peer.first, peer.second);
  1220. }
  1221. }
  1222. for (auto& peer : peers_to_drop)
  1223. DropPeer(peer.first, peer.second.get());
  1224. #endif
  1225. }
  1226. NodeController::IsolatedConnection::IsolatedConnection() = default;
  1227. NodeController::IsolatedConnection::IsolatedConnection(
  1228. const IsolatedConnection& other) = default;
  1229. NodeController::IsolatedConnection::IsolatedConnection(
  1230. IsolatedConnection&& other) = default;
  1231. NodeController::IsolatedConnection::IsolatedConnection(
  1232. scoped_refptr<NodeChannel> channel,
  1233. const ports::PortRef& local_port,
  1234. base::StringPiece name)
  1235. : channel(std::move(channel)), local_port(local_port), name(name) {}
  1236. NodeController::IsolatedConnection::~IsolatedConnection() = default;
  1237. NodeController::IsolatedConnection&
  1238. NodeController::IsolatedConnection::operator=(const IsolatedConnection& other) =
  1239. default;
  1240. NodeController::IsolatedConnection&
  1241. NodeController::IsolatedConnection::operator=(IsolatedConnection&& other) =
  1242. default;
  1243. BoundedPeerSet::BoundedPeerSet() = default;
  1244. BoundedPeerSet::~BoundedPeerSet() = default;
  1245. void BoundedPeerSet::Insert(const ports::NodeName& name) {
  1246. if (new_set_.size() == kHalfSize) {
  1247. old_set_.clear();
  1248. std::swap(old_set_, new_set_);
  1249. }
  1250. new_set_.insert(name);
  1251. }
  1252. bool BoundedPeerSet::Contains(const ports::NodeName& name) {
  1253. if (old_set_.find(name) != old_set_.end())
  1254. return true;
  1255. if (new_set_.find(name) != new_set_.end())
  1256. return true;
  1257. return false;
  1258. }
  1259. } // namespace core
  1260. } // namespace mojo