1 /*
2 * Copyright (c) 2022 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "remote_command_executor.h"
17 #ifdef EVENTHANDLER_ENABLE
18 #include "access_event_handler.h"
19 #endif
20 #include "constant_common.h"
21 #include "device_info_manager.h"
22 #include "singleton.h"
23 #include "soft_bus_channel.h"
24 #include "soft_bus_manager.h"
25 #include "token_sync_manager_service.h"
26
27 namespace OHOS {
28 namespace Security {
29 namespace AccessToken {
30 namespace {
31 static const std::string TASK_NAME = "RemoteCommandExecutor::ProcessBufferedCommandsWithThread";
32 } // namespace
RemoteCommandExecutor(const std::string & targetNodeId)33 RemoteCommandExecutor::RemoteCommandExecutor(const std::string &targetNodeId)
34 : targetNodeId_(targetNodeId), ptrChannel_(nullptr), mutex_(), commands_(), running_(false)
35 {
36 LOGD(ATM_DOMAIN, ATM_TAG, "RemoteCommandExecutor()");
37 }
38
~RemoteCommandExecutor()39 RemoteCommandExecutor::~RemoteCommandExecutor()
40 {
41 LOGD(ATM_DOMAIN, ATM_TAG, "~RemoteCommandExecutor() begin");
42 running_ = false;
43 }
44
CreateChannel(const std::string & targetNodeId)45 const std::shared_ptr<RpcChannel> RemoteCommandExecutor::CreateChannel(const std::string &targetNodeId)
46 {
47 LOGD(ATM_DOMAIN, ATM_TAG, "CreateChannel: targetNodeId=%{public}s",
48 ConstantCommon::EncryptDevId(targetNodeId).c_str());
49 // only consider SoftBusChannel
50 std::shared_ptr<RpcChannel> ptrChannel = std::make_shared<SoftBusChannel>(targetNodeId);
51 return ptrChannel;
52 }
53
54 /*
55 * called by RemoteCommandExecutor, RemoteCommandManager
56 */
ProcessOneCommand(const std::shared_ptr<BaseRemoteCommand> & ptrCommand)57 int RemoteCommandExecutor::ProcessOneCommand(const std::shared_ptr<BaseRemoteCommand>& ptrCommand)
58 {
59 if (ptrCommand == nullptr) {
60 LOGW(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, attempt to process on null command.",
61 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
62 return Constant::SUCCESS;
63 }
64 const std::string uniqueId = ptrCommand->remoteProtocol_.uniqueId;
65 LOGI(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, process one command start, uniqueId: %{public}s",
66 ConstantCommon::EncryptDevId(targetNodeId_).c_str(), uniqueId.c_str());
67
68 ptrCommand->Prepare();
69 int status = ptrCommand->remoteProtocol_.statusCode;
70 if (status != Constant::SUCCESS) {
71 LOGE(ATM_DOMAIN, ATM_TAG,
72 "targetNodeId %{public}s, process one command error, uniqueId: %{public}s, message: "
73 "prepare failure code %{public}d", ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
74 uniqueId.c_str(), status);
75 return status;
76 }
77
78 std::string localUdid = ConstantCommon::GetLocalDeviceId();
79 if (targetNodeId_ == localUdid) {
80 return ExecuteRemoteCommand(ptrCommand, false);
81 }
82
83 // otherwise a remote device
84 CreateChannelIfNeeded();
85 if (ptrChannel_ == nullptr) {
86 LOGE(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, channel is null.",
87 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
88 return Constant::FAILURE;
89 }
90 if (ptrChannel_->BuildConnection() != Constant::SUCCESS) {
91 LOGE(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, channel is not ready.",
92 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
93 return Constant::FAILURE;
94 }
95
96 return ExecuteRemoteCommand(ptrCommand, true);
97 }
98
99 /*
100 * called by RemoteCommandManager
101 */
AddCommand(const std::shared_ptr<BaseRemoteCommand> & ptrCommand)102 int RemoteCommandExecutor::AddCommand(const std::shared_ptr<BaseRemoteCommand>& ptrCommand)
103 {
104 if (ptrCommand == nullptr) {
105 LOGD(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, attempt to add an empty command.",
106 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
107 return Constant::INVALID_COMMAND;
108 }
109
110 const std::string uniqueId = ptrCommand->remoteProtocol_.uniqueId;
111 LOGD(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, add uniqueId %{public}s",
112 ConstantCommon::EncryptDevId(targetNodeId_).c_str(), uniqueId.c_str());
113
114 std::unique_lock<std::recursive_mutex> lock(mutex_);
115
116 // make sure do not have the same command in the command buffer
117 if (std::any_of(commands_.begin(), commands_.end(),
118 [uniqueId](const auto& buffCommand) {return buffCommand->remoteProtocol_.uniqueId == uniqueId; })) {
119 LOGW(ATM_DOMAIN, ATM_TAG,
120 "targetNodeId %{public}s, add uniqueId %{public}s, already exist in the buffer, skip",
121 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
122 uniqueId.c_str());
123 return Constant::SUCCESS;
124 }
125
126 commands_.push_back(ptrCommand);
127 return Constant::SUCCESS;
128 }
129
130 /*
131 * called by RemoteCommandExecutor.ProcessCommandThread, RemoteCommandManager
132 */
ProcessBufferedCommands(bool standalone)133 int RemoteCommandExecutor::ProcessBufferedCommands(bool standalone)
134 {
135 LOGI(ATM_DOMAIN, ATM_TAG, "Begin, targetNodeId: %{public}s, standalone: %{public}d",
136 ConstantCommon::EncryptDevId(targetNodeId_).c_str(), standalone);
137
138 std::unique_lock<std::recursive_mutex> lock(mutex_);
139
140 if (commands_.empty()) {
141 LOGW(ATM_DOMAIN, ATM_TAG, "No command, targetNodeId %{public}s",
142 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
143 running_ = false;
144 return Constant::SUCCESS;
145 }
146
147 running_ = true;
148 while (true) {
149 // interrupt
150 if (!running_) {
151 LOGI(ATM_DOMAIN, ATM_TAG, "End with running flag == false, targetNodeId: %{public}s",
152 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
153 return Constant::FAILURE;
154 }
155 // end
156 if (commands_.empty()) {
157 running_ = false;
158 LOGI(ATM_DOMAIN, ATM_TAG, "End, no command left, targetNodeId: %{public}s",
159 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
160 return Constant::SUCCESS;
161 }
162
163 // consume queue to execute
164 const std::shared_ptr<BaseRemoteCommand> bufferedCommand = commands_.front();
165 int status = ProcessOneCommand(bufferedCommand);
166 if (status == Constant::SUCCESS) {
167 commands_.pop_front();
168 continue;
169 } else if (status == Constant::FAILURE_BUT_CAN_RETRY) {
170 LOGW(ATM_DOMAIN, ATM_TAG,
171 "execute failed and wait to retry, targetNodeId: %{public}s, message: %{public}s, and will retry ",
172 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
173 bufferedCommand->remoteProtocol_.message.c_str());
174
175 // now, the retry at once will have no effective because the network problem
176 // so if the before the step, one command is added, and run this function
177 // it should also not need to restart to process the commands buffer at once.
178 running_ = false;
179 return Constant::FAILURE;
180 } else {
181 // this command failed, move on to execute next command
182 commands_.pop_front();
183 LOGE(ATM_DOMAIN, ATM_TAG,
184 "execute failed, targetNodeId: %{public}s, commandName: %{public}s, message: %{public}s",
185 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
186 bufferedCommand->remoteProtocol_.commandName.c_str(),
187 bufferedCommand->remoteProtocol_.message.c_str());
188 }
189 }
190 }
191
192 /*
193 * called by RemoteCommandManager
194 */
ProcessBufferedCommandsWithThread()195 void RemoteCommandExecutor::ProcessBufferedCommandsWithThread()
196 {
197 LOGI(ATM_DOMAIN, ATM_TAG, "Begin, targetNodeId: %{public}s", ConstantCommon::EncryptDevId(targetNodeId_).c_str());
198
199 std::unique_lock<std::recursive_mutex> lock(mutex_);
200
201 if (commands_.empty()) {
202 LOGI(ATM_DOMAIN, ATM_TAG, "No buffered commands. targetNodeId: %{public}s",
203 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
204 return;
205 }
206 if (running_) {
207 // task is running, do not need to start one more
208 LOGW(ATM_DOMAIN, ATM_TAG, "Task busy. targetNodeId: %{public}s",
209 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
210 return;
211 }
212
213 running_ = true;
214 const std::function<void()> runner = [weak = weak_from_this()]() {
215 auto self = weak.lock();
216 if (self == nullptr) {
217 LOGE(ATM_DOMAIN, ATM_TAG, "RemoteCommandExecutor is nullptr");
218 return;
219 }
220 self->ProcessBufferedCommands(true);
221 };
222
223 #ifdef EVENTHANDLER_ENABLE
224 auto tokenSyncManagerService = DelayedSingleton<TokenSyncManagerService>::GetInstance();
225 if (tokenSyncManagerService == nullptr) {
226 LOGE(ATM_DOMAIN, ATM_TAG, "TokenSyncManagerService is null.");
227 return;
228 }
229 std::shared_ptr<AccessEventHandler> handler = tokenSyncManagerService->GetSendEventHandler();
230 if (handler == nullptr) {
231 LOGE(ATM_DOMAIN, ATM_TAG, "Fail to get EventHandler");
232 return;
233 }
234 bool result = handler->ProxyPostTask(runner, TASK_NAME);
235 if (!result) {
236 LOGE(ATM_DOMAIN, ATM_TAG, "Post task failed, targetNodeId: %{public}s",
237 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
238 }
239 #endif
240 LOGI(ATM_DOMAIN, ATM_TAG,
241 "post task succeed, targetNodeId: %{public}s, taskName: %{public}s",
242 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
243 TASK_NAME.c_str());
244 }
245
ExecuteRemoteCommand(const std::shared_ptr<BaseRemoteCommand> & ptrCommand,const bool isRemote)246 int RemoteCommandExecutor::ExecuteRemoteCommand(
247 const std::shared_ptr<BaseRemoteCommand>& ptrCommand, const bool isRemote)
248 {
249 std::string uniqueId = ptrCommand->remoteProtocol_.uniqueId;
250 std::string tartgetNodeId = ConstantCommon::EncryptDevId(targetNodeId_);
251 LOGI(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, uniqueId %{public}s, remote %{public}d: start to execute.",
252 tartgetNodeId.c_str(), uniqueId.c_str(), isRemote);
253
254 ptrCommand->remoteProtocol_.statusCode = Constant::STATUS_CODE_BEFORE_RPC;
255
256 if (!isRemote) {
257 // Local device, play myself.
258 ptrCommand->Execute();
259 int code = ClientProcessResult(ptrCommand);
260 LOGD(ATM_DOMAIN, ATM_TAG, "Command finished with status: %{public}d, message: %{public}s.",
261 ptrCommand->remoteProtocol_.statusCode, ptrCommand->remoteProtocol_.message.c_str());
262 return code;
263 }
264
265 LOGI(ATM_DOMAIN, ATM_TAG, "Command executed uniqueId %{public}s.", uniqueId.c_str());
266
267 std::string responseString;
268 int32_t repeatTimes = SoftBusManager::GetInstance().GetRepeatTimes(); // repeat 5 times if responseString empty
269 if (ptrChannel_ == nullptr) {
270 LOGE(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, channel is null.", tartgetNodeId.c_str());
271 return Constant::FAILURE;
272 }
273 std::string commandName = ptrCommand->remoteProtocol_.commandName;
274 for (int32_t i = 0; i < repeatTimes; ++i) {
275 responseString = ptrChannel_->ExecuteCommand(commandName, ptrCommand->ToJsonPayload());
276 if (!responseString.empty()) {
277 break; // when responseString is not empty, break the loop
278 }
279
280 LOGW(ATM_DOMAIN, ATM_TAG,
281 "TargetNodeId %{public}s, uniqueId %{public}s, execute remote command error, response is empty.",
282 tartgetNodeId.c_str(), uniqueId.c_str());
283 }
284
285 if (responseString.empty()) {
286 if (commands_.empty()) {
287 ptrChannel_->CloseConnection(); // if command send failed, also try to close session
288 }
289 return Constant::FAILURE;
290 }
291
292 std::shared_ptr<BaseRemoteCommand> ptrResponseCommand =
293 RemoteCommandFactory::GetInstance().NewRemoteCommandFromJson(commandName, responseString);
294 if (ptrResponseCommand == nullptr) {
295 LOGE(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, get null response command!", tartgetNodeId.c_str());
296 return Constant::FAILURE;
297 }
298 int32_t result = ClientProcessResult(ptrResponseCommand);
299 if (commands_.empty()) {
300 ptrChannel_->CloseConnection();
301 }
302 LOGD(ATM_DOMAIN, ATM_TAG, "Command finished with status: %{public}d, message: %{public}s.",
303 ptrResponseCommand->remoteProtocol_.statusCode, ptrResponseCommand->remoteProtocol_.message.c_str());
304 return result;
305 }
306
CreateChannelIfNeeded()307 void RemoteCommandExecutor::CreateChannelIfNeeded()
308 {
309 std::unique_lock<std::recursive_mutex> lock(mutex_);
310 if (ptrChannel_ != nullptr) {
311 LOGI(ATM_DOMAIN, ATM_TAG, "TargetNodeId %{public}s, channel is exist.",
312 ConstantCommon::EncryptDevId(targetNodeId_).c_str());
313 return;
314 }
315
316 ptrChannel_ = CreateChannel(targetNodeId_);
317 }
318
ClientProcessResult(const std::shared_ptr<BaseRemoteCommand> & ptrCommand)319 int RemoteCommandExecutor::ClientProcessResult(const std::shared_ptr<BaseRemoteCommand>& ptrCommand)
320 {
321 std::string uniqueId = ptrCommand->remoteProtocol_.uniqueId;
322 if (ptrCommand->remoteProtocol_.statusCode == Constant::STATUS_CODE_BEFORE_RPC) {
323 LOGE(ATM_DOMAIN, ATM_TAG,
324 "targetNodeId %{public}s, uniqueId %{public}s, status code after RPC is same as before, the remote side "
325 "may not "
326 "support this command",
327 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
328 uniqueId.c_str());
329 return Constant::FAILURE;
330 }
331
332 ptrCommand->Finish();
333 int status = ptrCommand->remoteProtocol_.statusCode;
334 if (status != Constant::SUCCESS) {
335 LOGE(ATM_DOMAIN, ATM_TAG,
336 "targetNodeId %{public}s, uniqueId %{public}s, execute failed, message: %{public}s",
337 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
338 uniqueId.c_str(),
339 ptrCommand->remoteProtocol_.message.c_str());
340 } else {
341 LOGI(ATM_DOMAIN, ATM_TAG,
342 "targetNodeId %{public}s, uniqueId %{public}s, execute succeed.",
343 ConstantCommon::EncryptDevId(targetNodeId_).c_str(),
344 uniqueId.c_str());
345 }
346 return status;
347 }
348 } // namespace AccessToken
349 } // namespace Security
350 } // namespace OHOS
351