1 /*
2 * Copyright (c) 2021 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 "ability_sync.h"
17
18 #include "message_transform.h"
19 #include "version.h"
20 #include "db_errno.h"
21 #include "log_print.h"
22 #include "sync_types.h"
23 #include "db_common.h"
24 #include "single_ver_kvdb_sync_interface.h"
25 #include "single_ver_sync_task_context.h"
26 #include "single_ver_kv_sync_task_context.h"
27 #ifdef RELATIONAL_STORE
28 #include "relational_db_sync_interface.h"
29 #include "single_ver_relational_sync_task_context.h"
30 #endif
31
32 namespace DistributedDB {
AbilitySyncRequestPacket()33 AbilitySyncRequestPacket::AbilitySyncRequestPacket()
34 : protocolVersion_(ABILITY_SYNC_VERSION_V1),
35 sendCode_(E_OK),
36 softwareVersion_(SOFTWARE_VERSION_CURRENT),
37 secLabel_(0),
38 secFlag_(0),
39 schemaType_(0),
40 dbCreateTime_(0),
41 schemaVersion_(0)
42 {
43 }
44
~AbilitySyncRequestPacket()45 AbilitySyncRequestPacket::~AbilitySyncRequestPacket()
46 {
47 }
48
SetProtocolVersion(uint32_t protocolVersion)49 void AbilitySyncRequestPacket::SetProtocolVersion(uint32_t protocolVersion)
50 {
51 protocolVersion_ = protocolVersion;
52 }
53
GetProtocolVersion() const54 uint32_t AbilitySyncRequestPacket::GetProtocolVersion() const
55 {
56 return protocolVersion_;
57 }
58
SetSendCode(int32_t sendCode)59 void AbilitySyncRequestPacket::SetSendCode(int32_t sendCode)
60 {
61 sendCode_ = sendCode;
62 }
63
GetSendCode() const64 int32_t AbilitySyncRequestPacket::GetSendCode() const
65 {
66 return sendCode_;
67 }
68
SetSoftwareVersion(uint32_t swVersion)69 void AbilitySyncRequestPacket::SetSoftwareVersion(uint32_t swVersion)
70 {
71 softwareVersion_ = swVersion;
72 }
73
GetSoftwareVersion() const74 uint32_t AbilitySyncRequestPacket::GetSoftwareVersion() const
75 {
76 return softwareVersion_;
77 }
78
SetSchema(const std::string & schema)79 void AbilitySyncRequestPacket::SetSchema(const std::string &schema)
80 {
81 schema_ = schema;
82 }
83
GetSchema() const84 std::string AbilitySyncRequestPacket::GetSchema() const
85 {
86 return schema_;
87 }
88
SetSchemaType(uint32_t schemaType)89 void AbilitySyncRequestPacket::SetSchemaType(uint32_t schemaType)
90 {
91 schemaType_ = schemaType;
92 }
93
GetSchemaType() const94 uint32_t AbilitySyncRequestPacket::GetSchemaType() const
95 {
96 return schemaType_;
97 }
98
SetSecLabel(int32_t secLabel)99 void AbilitySyncRequestPacket::SetSecLabel(int32_t secLabel)
100 {
101 secLabel_ = secLabel;
102 }
103
GetSecLabel() const104 int32_t AbilitySyncRequestPacket::GetSecLabel() const
105 {
106 return secLabel_;
107 }
108
SetSecFlag(int32_t secFlag)109 void AbilitySyncRequestPacket::SetSecFlag(int32_t secFlag)
110 {
111 secFlag_ = secFlag;
112 }
113
GetSecFlag() const114 int32_t AbilitySyncRequestPacket::GetSecFlag() const
115 {
116 return secFlag_;
117 }
118
SetDbCreateTime(uint64_t dbCreateTime)119 void AbilitySyncRequestPacket::SetDbCreateTime(uint64_t dbCreateTime)
120 {
121 dbCreateTime_ = dbCreateTime;
122 }
123
GetDbCreateTime() const124 uint64_t AbilitySyncRequestPacket::GetDbCreateTime() const
125 {
126 return dbCreateTime_;
127 }
128
CalculateLen() const129 uint32_t AbilitySyncRequestPacket::CalculateLen() const
130 {
131 uint64_t len = 0;
132 len += Parcel::GetUInt32Len(); // protocolVersion_
133 len += Parcel::GetIntLen(); // sendCode_
134 len += Parcel::GetUInt32Len(); // softwareVersion_
135 uint32_t schemaLen = Parcel::GetStringLen(schema_);
136 if (schemaLen == 0) {
137 LOGE("[AbilitySyncRequestPacket][CalculateLen] schemaLen err!");
138 return 0;
139 }
140 len += schemaLen;
141 len += Parcel::GetIntLen(); // secLabel_
142 len += Parcel::GetIntLen(); // secFlag_
143 len += Parcel::GetUInt32Len(); // schemaType_
144 len += Parcel::GetUInt64Len(); // dbCreateTime_
145 len += DbAbility::CalculateLen(dbAbility_); // dbAbility_
146 len += Parcel::GetUInt64Len(); // schema version add in 109
147 // the reason why not 8-byte align is that old version is not 8-byte align
148 // so it is not possible to set 8-byte align for high version.
149 if (len > INT32_MAX) {
150 LOGE("[AbilitySyncRequestPacket][CalculateLen] err len:%" PRIu64, len);
151 return 0;
152 }
153 return len;
154 }
155
GetDbAbility() const156 DbAbility AbilitySyncRequestPacket::GetDbAbility() const
157 {
158 return dbAbility_;
159 }
160
SetDbAbility(const DbAbility & dbAbility)161 void AbilitySyncRequestPacket::SetDbAbility(const DbAbility &dbAbility)
162 {
163 dbAbility_ = dbAbility;
164 }
165
SetSchemaVersion(uint64_t schemaVersion)166 void AbilitySyncRequestPacket::SetSchemaVersion(uint64_t schemaVersion)
167 {
168 schemaVersion_ = schemaVersion;
169 }
170
GetSchemaVersion() const171 uint64_t AbilitySyncRequestPacket::GetSchemaVersion() const
172 {
173 return schemaVersion_;
174 }
175
AbilitySyncAckPacket()176 AbilitySyncAckPacket::AbilitySyncAckPacket()
177 : protocolVersion_(ABILITY_SYNC_VERSION_V1),
178 softwareVersion_(SOFTWARE_VERSION_CURRENT),
179 ackCode_(E_OK),
180 secLabel_(0),
181 secFlag_(0),
182 schemaType_(0),
183 permitSync_(0),
184 requirePeerConvert_(0),
185 dbCreateTime_(0),
186 schemaVersion_(0)
187 {
188 }
189
~AbilitySyncAckPacket()190 AbilitySyncAckPacket::~AbilitySyncAckPacket()
191 {
192 }
193
SetProtocolVersion(uint32_t protocolVersion)194 void AbilitySyncAckPacket::SetProtocolVersion(uint32_t protocolVersion)
195 {
196 protocolVersion_ = protocolVersion;
197 }
198
SetSoftwareVersion(uint32_t swVersion)199 void AbilitySyncAckPacket::SetSoftwareVersion(uint32_t swVersion)
200 {
201 softwareVersion_ = swVersion;
202 }
203
GetSoftwareVersion() const204 uint32_t AbilitySyncAckPacket::GetSoftwareVersion() const
205 {
206 return softwareVersion_;
207 }
208
GetProtocolVersion() const209 uint32_t AbilitySyncAckPacket::GetProtocolVersion() const
210 {
211 return protocolVersion_;
212 }
213
SetAckCode(int32_t ackCode)214 void AbilitySyncAckPacket::SetAckCode(int32_t ackCode)
215 {
216 ackCode_ = ackCode;
217 }
218
GetAckCode() const219 int32_t AbilitySyncAckPacket::GetAckCode() const
220 {
221 return ackCode_;
222 }
223
SetSchema(const std::string & schema)224 void AbilitySyncAckPacket::SetSchema(const std::string &schema)
225 {
226 schema_ = schema;
227 }
228
GetSchema() const229 std::string AbilitySyncAckPacket::GetSchema() const
230 {
231 return schema_;
232 }
233
SetSchemaType(uint32_t schemaType)234 void AbilitySyncAckPacket::SetSchemaType(uint32_t schemaType)
235 {
236 schemaType_ = schemaType;
237 }
238
GetSchemaType() const239 uint32_t AbilitySyncAckPacket::GetSchemaType() const
240 {
241 return schemaType_;
242 }
243
SetSecLabel(int32_t secLabel)244 void AbilitySyncAckPacket::SetSecLabel(int32_t secLabel)
245 {
246 secLabel_ = secLabel;
247 }
248
GetSecLabel() const249 int32_t AbilitySyncAckPacket::GetSecLabel() const
250 {
251 return secLabel_;
252 }
253
SetSecFlag(int32_t secFlag)254 void AbilitySyncAckPacket::SetSecFlag(int32_t secFlag)
255 {
256 secFlag_ = secFlag;
257 }
258
GetSecFlag() const259 int32_t AbilitySyncAckPacket::GetSecFlag() const
260 {
261 return secFlag_;
262 }
263
SetPermitSync(uint32_t permitSync)264 void AbilitySyncAckPacket::SetPermitSync(uint32_t permitSync)
265 {
266 permitSync_ = permitSync;
267 }
268
GetPermitSync() const269 uint32_t AbilitySyncAckPacket::GetPermitSync() const
270 {
271 return permitSync_;
272 }
273
SetRequirePeerConvert(uint32_t requirePeerConvert)274 void AbilitySyncAckPacket::SetRequirePeerConvert(uint32_t requirePeerConvert)
275 {
276 requirePeerConvert_ = requirePeerConvert;
277 }
278
GetRequirePeerConvert() const279 uint32_t AbilitySyncAckPacket::GetRequirePeerConvert() const
280 {
281 return requirePeerConvert_;
282 }
283
SetDbCreateTime(uint64_t dbCreateTime)284 void AbilitySyncAckPacket::SetDbCreateTime(uint64_t dbCreateTime)
285 {
286 dbCreateTime_ = dbCreateTime;
287 }
288
GetDbCreateTime() const289 uint64_t AbilitySyncAckPacket::GetDbCreateTime() const
290 {
291 return dbCreateTime_;
292 }
293
GetSchemaVersion() const294 uint64_t AbilitySyncAckPacket::GetSchemaVersion() const
295 {
296 return schemaVersion_;
297 }
298
SetSchemaVersion(uint64_t schemaVersion)299 void AbilitySyncAckPacket::SetSchemaVersion(uint64_t schemaVersion)
300 {
301 schemaVersion_ = schemaVersion;
302 }
303
CalculateLen() const304 uint32_t AbilitySyncAckPacket::CalculateLen() const
305 {
306 uint64_t len = 0;
307 len += Parcel::GetUInt32Len();
308 len += Parcel::GetUInt32Len();
309 len += Parcel::GetIntLen();
310 uint32_t schemaLen = Parcel::GetStringLen(schema_);
311 if (schemaLen == 0) {
312 LOGE("[AbilitySyncAckPacket][CalculateLen] schemaLen err!");
313 return 0;
314 }
315 len += schemaLen;
316 len += Parcel::GetIntLen(); // secLabel_
317 len += Parcel::GetIntLen(); // secFlag_
318 len += Parcel::GetUInt32Len(); // schemaType_
319 len += Parcel::GetUInt32Len(); // permitSync_
320 len += Parcel::GetUInt32Len(); // requirePeerConvert_
321 len += Parcel::GetUInt64Len(); // dbCreateTime_
322 len += DbAbility::CalculateLen(dbAbility_); // dbAbility_
323 len += SchemaNegotiate::CalculateParcelLen(relationalSyncOpinion_);
324 len += Parcel::GetUInt64Len(); // schemaVersion_
325 if (len > INT32_MAX) {
326 LOGE("[AbilitySyncAckPacket][CalculateLen] err len:%" PRIu64, len);
327 return 0;
328 }
329 return len;
330 }
331
GetDbAbility() const332 DbAbility AbilitySyncAckPacket::GetDbAbility() const
333 {
334 return dbAbility_;
335 }
336
SetDbAbility(const DbAbility & dbAbility)337 void AbilitySyncAckPacket::SetDbAbility(const DbAbility &dbAbility)
338 {
339 dbAbility_ = dbAbility;
340 }
341
SetRelationalSyncOpinion(const RelationalSyncOpinion & relationalSyncOpinion)342 void AbilitySyncAckPacket::SetRelationalSyncOpinion(const RelationalSyncOpinion &relationalSyncOpinion)
343 {
344 relationalSyncOpinion_ = relationalSyncOpinion;
345 }
346
GetRelationalSyncOpinion() const347 RelationalSyncOpinion AbilitySyncAckPacket::GetRelationalSyncOpinion() const
348 {
349 return relationalSyncOpinion_;
350 }
351
AbilitySync()352 AbilitySync::AbilitySync()
353 : communicator_(nullptr),
354 storageInterface_(nullptr),
355 metadata_(nullptr),
356 syncFinished_(false)
357 {
358 }
359
~AbilitySync()360 AbilitySync::~AbilitySync()
361 {
362 communicator_ = nullptr;
363 storageInterface_ = nullptr;
364 metadata_ = nullptr;
365 }
366
Initialize(ICommunicator * inCommunicator,ISyncInterface * inStorage,const std::shared_ptr<Metadata> & inMetadata,const std::string & deviceId)367 int AbilitySync::Initialize(ICommunicator *inCommunicator, ISyncInterface *inStorage,
368 const std::shared_ptr<Metadata> &inMetadata, const std::string &deviceId)
369 {
370 if (inCommunicator == nullptr || inStorage == nullptr || deviceId.empty() || inMetadata == nullptr) {
371 return -E_INVALID_ARGS;
372 }
373 communicator_ = inCommunicator;
374 storageInterface_ = inStorage;
375 metadata_ = inMetadata;
376 deviceId_ = deviceId;
377 return E_OK;
378 }
379
SyncStart(uint32_t sessionId,uint32_t sequenceId,uint16_t remoteCommunicatorVersion,const CommErrHandler & handler,const ISyncTaskContext * context)380 int AbilitySync::SyncStart(uint32_t sessionId, uint32_t sequenceId, uint16_t remoteCommunicatorVersion,
381 const CommErrHandler &handler, const ISyncTaskContext *context)
382 {
383 AbilitySyncRequestPacket packet;
384 int errCode = SetAbilityRequestBodyInfo(remoteCommunicatorVersion, context, packet);
385 if (errCode != E_OK) {
386 return errCode;
387 }
388 Message *message = new (std::nothrow) Message(ABILITY_SYNC_MESSAGE);
389 if (message == nullptr) {
390 return -E_OUT_OF_MEMORY;
391 }
392 message->SetMessageType(TYPE_REQUEST);
393 errCode = message->SetCopiedObject<>(packet);
394 if (errCode != E_OK) {
395 LOGE("[AbilitySync][SyncStart] SetCopiedObject failed, err %d", errCode);
396 delete message;
397 message = nullptr;
398 return errCode;
399 }
400 message->SetVersion(MSG_VERSION_EXT);
401 message->SetSessionId(sessionId);
402 message->SetSequenceId(sequenceId);
403 SendConfig conf;
404 SetSendConfigParam(storageInterface_->GetDbProperties(), deviceId_, false, SEND_TIME_OUT, conf);
405 errCode = communicator_->SendMessage(deviceId_, message, conf, handler);
406 if (errCode != E_OK) {
407 LOGE("[AbilitySync][SyncStart] SendPacket failed, err %d", errCode);
408 delete message;
409 message = nullptr;
410 }
411 return errCode;
412 }
413
AckRecv(const Message * message,ISyncTaskContext * context)414 int AbilitySync::AckRecv(const Message *message, ISyncTaskContext *context)
415 {
416 int errCode = AckMsgCheck(message, context);
417 if (errCode != E_OK) {
418 return errCode;
419 }
420 const AbilitySyncAckPacket *packet = message->GetObject<AbilitySyncAckPacket>();
421 if (packet == nullptr) {
422 return -E_INVALID_ARGS;
423 }
424 uint32_t remoteSoftwareVersion = packet->GetSoftwareVersion();
425 context->SetRemoteSoftwareVersion(remoteSoftwareVersion);
426 metadata_->SetRemoteSchemaVersion(context->GetDeviceId(), remoteSoftwareVersion);
427 if (remoteSoftwareVersion > SOFTWARE_VERSION_RELEASE_2_0) {
428 errCode = AckRecvWithHighVersion(message, context, packet);
429 } else {
430 std::string schema = packet->GetSchema();
431 uint8_t schemaType = packet->GetSchemaType();
432 bool isCompatible = static_cast<SyncGenericInterface *>(storageInterface_)->CheckCompatible(schema, schemaType);
433 if (!isCompatible) { // LCOV_EXCL_BR_LINE
434 (static_cast<SingleVerSyncTaskContext *>(context))->SetTaskErrCode(-E_SCHEMA_MISMATCH);
435 LOGE("[AbilitySync][AckRecv] scheme check failed");
436 return -E_SCHEMA_MISMATCH;
437 }
438 LOGI("[AbilitySync][AckRecv]remoteSoftwareVersion = %u, isCompatible = %d,", remoteSoftwareVersion,
439 isCompatible);
440 }
441 return errCode;
442 }
443
RequestRecv(const Message * message,ISyncTaskContext * context)444 int AbilitySync::RequestRecv(const Message *message, ISyncTaskContext *context)
445 {
446 if (message == nullptr || context == nullptr) {
447 return -E_INVALID_ARGS;
448 }
449 const AbilitySyncRequestPacket *packet = message->GetObject<AbilitySyncRequestPacket>();
450 if (packet == nullptr) {
451 return -E_INVALID_ARGS;
452 }
453 if (packet->GetSendCode() == -E_VERSION_NOT_SUPPORT) {
454 AbilitySyncAckPacket ackPacket;
455 (void)SendAck(context, message, -E_VERSION_NOT_SUPPORT, false, ackPacket);
456 LOGI("[AbilitySync][RequestRecv] version can not support, remote version is %u", packet->GetProtocolVersion());
457 return -E_VERSION_NOT_SUPPORT;
458 }
459
460 std::string schema = packet->GetSchema();
461 uint8_t schemaType = packet->GetSchemaType();
462 bool isCompatible = static_cast<SyncGenericInterface *>(storageInterface_)->CheckCompatible(schema, schemaType);
463 if (!isCompatible) {
464 (static_cast<SingleVerSyncTaskContext *>(context))->SetTaskErrCode(-E_SCHEMA_MISMATCH);
465 }
466 uint32_t remoteSoftwareVersion = packet->GetSoftwareVersion();
467 context->SetRemoteSoftwareVersion(remoteSoftwareVersion);
468 return HandleRequestRecv(message, context, isCompatible);
469 }
470
AckNotifyRecv(const Message * message,ISyncTaskContext * context)471 int AbilitySync::AckNotifyRecv(const Message *message, ISyncTaskContext *context)
472 {
473 if (message == nullptr || context == nullptr) {
474 return -E_INVALID_ARGS;
475 }
476 if (message->GetErrorNo() == E_FEEDBACK_UNKNOWN_MESSAGE) {
477 LOGE("[AbilitySync][AckNotifyRecv] Remote device dose not support this message id");
478 context->SetRemoteSoftwareVersion(SOFTWARE_VERSION_EARLIEST);
479 return -E_FEEDBACK_UNKNOWN_MESSAGE;
480 }
481 const AbilitySyncAckPacket *packet = message->GetObject<AbilitySyncAckPacket>();
482 if (packet == nullptr) {
483 return -E_INVALID_ARGS;
484 }
485 int errCode = packet->GetAckCode();
486 if (errCode != E_OK) {
487 LOGE("[AbilitySync][AckNotifyRecv] received an errCode %d", errCode);
488 return errCode;
489 }
490 uint32_t remoteSoftwareVersion = packet->GetSoftwareVersion();
491 context->SetRemoteSoftwareVersion(remoteSoftwareVersion);
492 AbilitySyncAckPacket sendPacket;
493 std::pair<bool, bool> schemaSyncStatus;
494 errCode = HandleVersionV3AckSchemaParam(packet, sendPacket, context, false, schemaSyncStatus);
495 errCode = errCode == -E_ABILITY_SYNC_FINISHED ? E_OK : errCode;
496 int ackCode = errCode;
497 LOGI("[AckNotifyRecv] receive dev = %s ack notify, remoteSoftwareVersion = %u, ackCode = %d",
498 STR_MASK(deviceId_), remoteSoftwareVersion, errCode);
499 if (errCode == E_OK) {
500 ackCode = AbilitySync::LAST_NOTIFY;
501 }
502 (void)SendAckWithEmptySchema(context, message, ackCode, true);
503 return errCode;
504 }
505
GetAbilitySyncFinishedStatus() const506 bool AbilitySync::GetAbilitySyncFinishedStatus() const
507 {
508 return syncFinished_;
509 }
510
SetAbilitySyncFinishedStatus(bool syncFinished,ISyncTaskContext & context)511 void AbilitySync::SetAbilitySyncFinishedStatus(bool syncFinished, ISyncTaskContext &context)
512 {
513 syncFinished_ = syncFinished;
514 if (context.GetRemoteSoftwareVersion() < SOFTWARE_VERSION_RELEASE_9_0) {
515 return;
516 }
517 // record finished with all schema compatible
518 if (syncFinished && !context.IsSchemaCompatible()) { // LCOV_EXCL_BR_LINE
519 return;
520 }
521 int errCode = metadata_->SetAbilitySyncFinishMark(deviceId_, syncFinished);
522 if (errCode != E_OK) {
523 LOGW("[AbilitySync] Set ability sync finish mark failed %d", errCode);
524 }
525 }
526
SecLabelCheckInner(int32_t remoteSecLabel,int errCode,SecurityOption option,const AbilitySyncRequestPacket * packet) const527 bool AbilitySync::SecLabelCheckInner(int32_t remoteSecLabel, int errCode, SecurityOption option,
528 const AbilitySyncRequestPacket *packet) const
529 {
530 if (remoteSecLabel == NOT_SUPPORT_SEC_CLASSIFICATION || remoteSecLabel == SecurityLabel::NOT_SET) {
531 return true;
532 }
533 if (errCode == -E_NOT_SUPPORT || (errCode == E_OK && option.securityLabel == SecurityLabel::NOT_SET)) {
534 return true;
535 }
536 if (remoteSecLabel == FAILED_GET_SEC_CLASSIFICATION || errCode != E_OK) {
537 LOGE("[AbilitySync][RequestRecv] check error remoteL:%d, errCode:%d", remoteSecLabel, errCode);
538 return false;
539 }
540 if (remoteSecLabel == option.securityLabel) {
541 return true;
542 }
543 LOGE("[AbilitySync][RequestRecv] check error remote:%d , %d local:%d , %d",
544 remoteSecLabel, packet->GetSecFlag(), option.securityLabel, option.securityFlag);
545 return false;
546 }
547
SecLabelCheck(const AbilitySyncRequestPacket * packet) const548 bool AbilitySync::SecLabelCheck(const AbilitySyncRequestPacket *packet) const
549 {
550 SecurityOption option;
551 auto *syncInterface = static_cast<SyncGenericInterface *>(storageInterface_);
552 if (syncInterface == nullptr) {
553 LOGE("[AbilitySync][RequestRecv] get sync interface wrong");
554 return false;
555 }
556 int errCode = syncInterface->GetSecurityOption(option);
557 int32_t remoteSecLabel = TransformSecLabelIfNeed(packet->GetSecLabel(), option.securityLabel);
558 LOGI("[AbilitySync][RequestRecv] remote label:%d local l:%d, f:%d, errCode:%d", remoteSecLabel,
559 option.securityLabel, option.securityFlag, errCode);
560 if (remoteSecLabel == NOT_SUPPORT_SEC_CLASSIFICATION && errCode == -E_NOT_SUPPORT) {
561 return true;
562 }
563 uint32_t remoteSoftwareVersion = packet->GetSoftwareVersion();
564 if (errCode != -E_NOT_SUPPORT && option.securityLabel == SecurityLabel::NOT_SET) {
565 LOGE("[AbilitySync][RequestRecv] local security label not set!");
566 return false;
567 }
568 if (remoteSoftwareVersion >= SOFTWARE_VERSION_RELEASE_7_0 && remoteSecLabel == SecurityLabel::NOT_SET) {
569 LOGE("[AbilitySync][RequestRecv] remote security label not set!");
570 return false;
571 }
572 return SecLabelCheckInner(remoteSecLabel, errCode, option, packet);
573 }
574
HandleVersionV3RequestParam(const AbilitySyncRequestPacket * packet,ISyncTaskContext * context)575 void AbilitySync::HandleVersionV3RequestParam(const AbilitySyncRequestPacket *packet, ISyncTaskContext *context)
576 {
577 int32_t remoteSecLabel = packet->GetSecLabel();
578 int32_t remoteSecFlag = packet->GetSecFlag();
579 DbAbility remoteDbAbility = packet->GetDbAbility();
580 (static_cast<SingleVerSyncTaskContext *>(context))->SetDbAbility(remoteDbAbility);
581 (static_cast<SingleVerSyncTaskContext *>(context))->SetRemoteSeccurityOption({remoteSecLabel, remoteSecFlag});
582 (static_cast<SingleVerSyncTaskContext *>(context))->SetReceivcPermitCheck(false);
583 LOGI("[AbilitySync][HandleVersionV3RequestParam] remoteSecLabel = %d, remoteSecFlag = %d, remoteSchemaType = %u",
584 remoteSecLabel, remoteSecFlag, packet->GetSchemaType());
585 }
586
HandleVersionV3AckSecOptionParam(const AbilitySyncAckPacket * packet,ISyncTaskContext * context)587 void AbilitySync::HandleVersionV3AckSecOptionParam(const AbilitySyncAckPacket *packet,
588 ISyncTaskContext *context)
589 {
590 int32_t remoteSecLabel = packet->GetSecLabel();
591 int32_t remoteSecFlag = packet->GetSecFlag();
592 SecurityOption secOption = {remoteSecLabel, remoteSecFlag};
593 (static_cast<SingleVerSyncTaskContext *>(context))->SetRemoteSeccurityOption(secOption);
594 (static_cast<SingleVerSyncTaskContext *>(context))->SetSendPermitCheck(false);
595 LOGI("[AbilitySync][AckRecv] remoteSecLabel = %d, remoteSecFlag = %d", remoteSecLabel, remoteSecFlag);
596 }
597
HandleVersionV3AckSchemaParam(const AbilitySyncAckPacket * recvPacket,AbilitySyncAckPacket & sendPacket,ISyncTaskContext * context,bool sendOpinion,std::pair<bool,bool> & schemaSyncStatus)598 int AbilitySync::HandleVersionV3AckSchemaParam(const AbilitySyncAckPacket *recvPacket,
599 AbilitySyncAckPacket &sendPacket, ISyncTaskContext *context, bool sendOpinion,
600 std::pair<bool, bool> &schemaSyncStatus)
601 {
602 if (IsSingleRelationalVer()) {
603 return HandleRelationAckSchemaParam(recvPacket, sendPacket, context, sendOpinion, schemaSyncStatus);
604 }
605 return HandleKvAckSchemaParam(recvPacket, context, sendPacket, schemaSyncStatus);
606 }
607
GetPacketSecOption(const ISyncTaskContext * context,SecurityOption & option) const608 void AbilitySync::GetPacketSecOption(const ISyncTaskContext *context, SecurityOption &option) const
609 {
610 int errCode =
611 (static_cast<SyncGenericInterface *>(storageInterface_))->GetSecurityOption(option);
612 if (errCode == -E_NOT_SUPPORT) {
613 LOGE("[AbilitySync][SyncStart] GetSecOpt not surpport sec classification");
614 option.securityLabel = NOT_SUPPORT_SEC_CLASSIFICATION;
615 } else if (errCode != E_OK) {
616 LOGE("[AbilitySync][SyncStart] GetSecOpt errCode:%d", errCode);
617 option.securityLabel = FAILED_GET_SEC_CLASSIFICATION;
618 }
619 if (context == nullptr) {
620 return;
621 }
622 auto remoteSecOption = (static_cast<const SingleVerSyncTaskContext *>(context))->GetRemoteSeccurityOption();
623 option.securityLabel = TransformSecLabelIfNeed(option.securityLabel, remoteSecOption.securityLabel);
624 }
625
RegisterTransformFunc()626 int AbilitySync::RegisterTransformFunc()
627 {
628 TransformFunc func;
629 func.computeFunc = [](const Message *inMsg) { return CalculateLen(inMsg); };
630 func.serializeFunc = [](uint8_t *buffer, uint32_t length, const Message *inMsg) {
631 return Serialization(buffer, length, inMsg);
632 };
633 func.deserializeFunc = [](const uint8_t *buffer, uint32_t length, Message *inMsg) {
634 return DeSerialization(buffer, length, inMsg);
635 };
636 return MessageTransform::RegTransformFunction(ABILITY_SYNC_MESSAGE, func);
637 }
638
CalculateLen(const Message * inMsg)639 uint32_t AbilitySync::CalculateLen(const Message *inMsg)
640 {
641 if ((inMsg == nullptr) || (inMsg->GetMessageId() != ABILITY_SYNC_MESSAGE)) {
642 return 0;
643 }
644 int errCode;
645 uint32_t len = 0;
646 switch (inMsg->GetMessageType()) {
647 case TYPE_REQUEST:
648 errCode = RequestPacketCalculateLen(inMsg, len);
649 if (errCode != E_OK) {
650 LOGE("[AbilitySync][CalculateLen] request packet calc length err %d", errCode);
651 }
652 break;
653 case TYPE_RESPONSE:
654 errCode = AckPacketCalculateLen(inMsg, len);
655 if (errCode != E_OK) {
656 LOGE("[AbilitySync][CalculateLen] ack packet calc length err %d", errCode);
657 }
658 break;
659 case TYPE_NOTIFY:
660 errCode = AckPacketCalculateLen(inMsg, len);
661 if (errCode != E_OK) {
662 LOGE("[AbilitySync][CalculateLen] ack packet calc length err %d", errCode);
663 }
664 break;
665 default:
666 LOGE("[AbilitySync][CalculateLen] message type not support, type %d", inMsg->GetMessageType());
667 break;
668 }
669 return len;
670 }
671
Serialization(uint8_t * buffer,uint32_t length,const Message * inMsg)672 int AbilitySync::Serialization(uint8_t *buffer, uint32_t length, const Message *inMsg)
673 {
674 if ((buffer == nullptr) || (inMsg == nullptr)) {
675 return -E_INVALID_ARGS;
676 }
677
678 switch (inMsg->GetMessageType()) {
679 case TYPE_REQUEST:
680 return RequestPacketSerialization(buffer, length, inMsg);
681 case TYPE_RESPONSE:
682 case TYPE_NOTIFY:
683 return AckPacketSerialization(buffer, length, inMsg);
684 default:
685 return -E_MESSAGE_TYPE_ERROR;
686 }
687 }
688
DeSerialization(const uint8_t * buffer,uint32_t length,Message * inMsg)689 int AbilitySync::DeSerialization(const uint8_t *buffer, uint32_t length, Message *inMsg)
690 {
691 if ((buffer == nullptr) || (inMsg == nullptr)) {
692 return -E_INVALID_ARGS;
693 }
694
695 switch (inMsg->GetMessageType()) {
696 case TYPE_REQUEST:
697 return RequestPacketDeSerialization(buffer, length, inMsg);
698 case TYPE_RESPONSE:
699 case TYPE_NOTIFY:
700 return AckPacketDeSerialization(buffer, length, inMsg);
701 default:
702 return -E_MESSAGE_TYPE_ERROR;
703 }
704 }
705
RequestPacketCalculateLen(const Message * inMsg,uint32_t & len)706 int AbilitySync::RequestPacketCalculateLen(const Message *inMsg, uint32_t &len)
707 {
708 const AbilitySyncRequestPacket *packet = inMsg->GetObject<AbilitySyncRequestPacket>();
709 if (packet == nullptr) {
710 return -E_INVALID_ARGS;
711 }
712
713 len = packet->CalculateLen();
714 return E_OK;
715 }
716
AckPacketCalculateLen(const Message * inMsg,uint32_t & len)717 int AbilitySync::AckPacketCalculateLen(const Message *inMsg, uint32_t &len)
718 {
719 const AbilitySyncAckPacket *packet = inMsg->GetObject<AbilitySyncAckPacket>();
720 if (packet == nullptr) {
721 return -E_INVALID_ARGS;
722 }
723
724 len = packet->CalculateLen();
725 return E_OK;
726 }
727
RequestPacketSerialization(uint8_t * buffer,uint32_t length,const Message * inMsg)728 int AbilitySync::RequestPacketSerialization(uint8_t *buffer, uint32_t length, const Message *inMsg)
729 {
730 const AbilitySyncRequestPacket *packet = inMsg->GetObject<AbilitySyncRequestPacket>();
731 if ((packet == nullptr) || (length != packet->CalculateLen())) {
732 return -E_INVALID_ARGS;
733 }
734
735 Parcel parcel(buffer, length);
736 parcel.WriteUInt32(packet->GetProtocolVersion());
737 parcel.WriteInt(packet->GetSendCode());
738 parcel.WriteUInt32(packet->GetSoftwareVersion());
739 parcel.WriteString(packet->GetSchema());
740 parcel.WriteInt(packet->GetSecLabel());
741 parcel.WriteInt(packet->GetSecFlag());
742 parcel.WriteUInt32(packet->GetSchemaType());
743 parcel.WriteUInt64(packet->GetDbCreateTime());
744 int errCode = DbAbility::Serialize(parcel, packet->GetDbAbility());
745 parcel.WriteUInt64(packet->GetSchemaVersion());
746 if (parcel.IsError() || errCode != E_OK) {
747 return -E_PARSE_FAIL;
748 }
749 return E_OK;
750 }
751
AckPacketSerialization(uint8_t * buffer,uint32_t length,const Message * inMsg)752 int AbilitySync::AckPacketSerialization(uint8_t *buffer, uint32_t length, const Message *inMsg)
753 {
754 const AbilitySyncAckPacket *packet = inMsg->GetObject<AbilitySyncAckPacket>();
755 if ((packet == nullptr) || (length != packet->CalculateLen())) {
756 return -E_INVALID_ARGS;
757 }
758
759 Parcel parcel(buffer, length);
760 parcel.WriteUInt32(ABILITY_SYNC_VERSION_V1);
761 parcel.WriteUInt32(SOFTWARE_VERSION_CURRENT);
762 parcel.WriteInt(packet->GetAckCode());
763 parcel.WriteString(packet->GetSchema());
764 parcel.WriteInt(packet->GetSecLabel());
765 parcel.WriteInt(packet->GetSecFlag());
766 parcel.WriteUInt32(packet->GetSchemaType());
767 parcel.WriteUInt32(packet->GetPermitSync());
768 parcel.WriteUInt32(packet->GetRequirePeerConvert());
769 parcel.WriteUInt64(packet->GetDbCreateTime());
770 int errCode = DbAbility::Serialize(parcel, packet->GetDbAbility());
771 if (parcel.IsError() || errCode != E_OK) {
772 return -E_PARSE_FAIL;
773 }
774 errCode = SchemaNegotiate::SerializeData(packet->GetRelationalSyncOpinion(), parcel);
775 if (parcel.IsError() || errCode != E_OK) {
776 return -E_PARSE_FAIL;
777 }
778 parcel.WriteUInt64(packet->GetSchemaVersion());
779 if (parcel.IsError()) {
780 LOGE("[AbilitySync] Serialize schema version failed");
781 return -E_PARSE_FAIL;
782 }
783 return E_OK;
784 }
785
RequestPacketDeSerialization(const uint8_t * buffer,uint32_t length,Message * inMsg)786 int AbilitySync::RequestPacketDeSerialization(const uint8_t *buffer, uint32_t length, Message *inMsg)
787 {
788 auto *packet = new (std::nothrow) AbilitySyncRequestPacket();
789 if (packet == nullptr) {
790 return -E_OUT_OF_MEMORY;
791 }
792
793 Parcel parcel(const_cast<uint8_t *>(buffer), length);
794 uint32_t version = 0;
795 uint32_t softwareVersion = 0;
796 std::string schema;
797 int32_t sendCode = 0;
798 int errCode = -E_PARSE_FAIL;
799
800 parcel.ReadUInt32(version);
801 if (parcel.IsError()) {
802 goto ERROR_OUT;
803 }
804 packet->SetProtocolVersion(version);
805 if (version > ABILITY_SYNC_VERSION_V1) {
806 packet->SetSendCode(-E_VERSION_NOT_SUPPORT);
807 errCode = inMsg->SetExternalObject<>(packet);
808 if (errCode != E_OK) {
809 goto ERROR_OUT;
810 }
811 return errCode;
812 }
813 parcel.ReadInt(sendCode);
814 parcel.ReadUInt32(softwareVersion);
815 parcel.ReadString(schema);
816 errCode = RequestPacketDeSerializationTailPart(parcel, packet, softwareVersion);
817 if (parcel.IsError() || errCode != E_OK) {
818 goto ERROR_OUT;
819 }
820 packet->SetSendCode(sendCode);
821 packet->SetSoftwareVersion(softwareVersion);
822 packet->SetSchema(schema);
823
824 errCode = inMsg->SetExternalObject<>(packet);
825 if (errCode == E_OK) {
826 return E_OK;
827 }
828
829 ERROR_OUT:
830 delete packet;
831 return errCode;
832 }
833
RequestPacketDeSerializationTailPart(Parcel & parcel,AbilitySyncRequestPacket * packet,uint32_t version)834 int AbilitySync::RequestPacketDeSerializationTailPart(Parcel &parcel, AbilitySyncRequestPacket *packet,
835 uint32_t version)
836 {
837 if (!parcel.IsError() && version > SOFTWARE_VERSION_RELEASE_2_0) {
838 int32_t secLabel = 0;
839 int32_t secFlag = 0;
840 uint32_t schemaType = 0;
841 parcel.ReadInt(secLabel);
842 parcel.ReadInt(secFlag);
843 parcel.ReadUInt32(schemaType);
844 packet->SetSecLabel(secLabel);
845 packet->SetSecFlag(secFlag);
846 packet->SetSchemaType(schemaType);
847 }
848 if (!parcel.IsError() && version > SOFTWARE_VERSION_RELEASE_3_0) {
849 uint64_t dbCreateTime = 0;
850 parcel.ReadUInt64(dbCreateTime);
851 packet->SetDbCreateTime(dbCreateTime);
852 }
853 DbAbility remoteDbAbility;
854 int errCode = DbAbility::DeSerialize(parcel, remoteDbAbility);
855 if (errCode != E_OK) {
856 LOGE("[AbilitySync] request packet DeSerializ failed.");
857 return errCode;
858 }
859 packet->SetDbAbility(remoteDbAbility);
860 if (version >= SOFTWARE_VERSION_RELEASE_9_0) {
861 uint64_t schemaVersion = 0u;
862 parcel.ReadUInt64(schemaVersion);
863 if (parcel.IsError()) {
864 LOGW("[AbilitySync] request packet read schema version failed");
865 return -E_PARSE_FAIL;
866 }
867 packet->SetSchemaVersion(schemaVersion);
868 }
869 return E_OK;
870 }
871
AckPacketDeSerializationTailPart(Parcel & parcel,AbilitySyncAckPacket * packet,uint32_t version)872 int AbilitySync::AckPacketDeSerializationTailPart(Parcel &parcel, AbilitySyncAckPacket *packet, uint32_t version)
873 {
874 if (!parcel.IsError() && version > SOFTWARE_VERSION_RELEASE_2_0) {
875 int32_t secLabel = 0;
876 int32_t secFlag = 0;
877 uint32_t schemaType = 0;
878 uint32_t permitSync = 0;
879 uint32_t requirePeerConvert = 0;
880 parcel.ReadInt(secLabel);
881 parcel.ReadInt(secFlag);
882 parcel.ReadUInt32(schemaType);
883 parcel.ReadUInt32(permitSync);
884 parcel.ReadUInt32(requirePeerConvert);
885 packet->SetSecLabel(secLabel);
886 packet->SetSecFlag(secFlag);
887 packet->SetSchemaType(schemaType);
888 packet->SetPermitSync(permitSync);
889 packet->SetRequirePeerConvert(requirePeerConvert);
890 }
891 if (!parcel.IsError() && version > SOFTWARE_VERSION_RELEASE_3_0) {
892 uint64_t dbCreateTime = 0;
893 parcel.ReadUInt64(dbCreateTime);
894 packet->SetDbCreateTime(dbCreateTime);
895 }
896 DbAbility remoteDbAbility;
897 int errCode = DbAbility::DeSerialize(parcel, remoteDbAbility);
898 if (errCode != E_OK) {
899 LOGE("[AbilitySync] ack packet DeSerializ failed.");
900 return errCode;
901 }
902 packet->SetDbAbility(remoteDbAbility);
903 RelationalSyncOpinion relationalSyncOpinion;
904 errCode = SchemaNegotiate::DeserializeData(parcel, relationalSyncOpinion);
905 if (errCode != E_OK) {
906 LOGE("[AbilitySync] ack packet DeSerializ RelationalSyncOpinion failed.");
907 return errCode;
908 }
909 packet->SetRelationalSyncOpinion(relationalSyncOpinion);
910 if (version >= SOFTWARE_VERSION_RELEASE_9_0) {
911 uint64_t schemaVersion = 0;
912 parcel.ReadUInt64(schemaVersion);
913 if (parcel.IsError()) {
914 LOGW("[AbilitySync] ack packet read schema version failed.");
915 return -E_PARSE_FAIL;
916 }
917 packet->SetSchemaVersion(schemaVersion);
918 }
919 return E_OK;
920 }
921
AckPacketDeSerialization(const uint8_t * buffer,uint32_t length,Message * inMsg)922 int AbilitySync::AckPacketDeSerialization(const uint8_t *buffer, uint32_t length, Message *inMsg)
923 {
924 auto *packet = new (std::nothrow) AbilitySyncAckPacket();
925 if (packet == nullptr) {
926 return -E_OUT_OF_MEMORY;
927 }
928
929 Parcel parcel(const_cast<uint8_t *>(buffer), length);
930 uint32_t version = 0;
931 uint32_t softwareVersion = 0;
932 int32_t ackCode = E_OK;
933 std::string schema;
934 int errCode;
935 parcel.ReadUInt32(version);
936 if (parcel.IsError()) {
937 LOGE("[AbilitySync][RequestDeSerialization] read version failed!");
938 errCode = -E_PARSE_FAIL;
939 goto ERROR_OUT;
940 }
941 packet->SetProtocolVersion(version);
942 if (version > ABILITY_SYNC_VERSION_V1) {
943 packet->SetAckCode(-E_VERSION_NOT_SUPPORT);
944 errCode = inMsg->SetExternalObject<>(packet);
945 if (errCode != E_OK) {
946 goto ERROR_OUT;
947 }
948 return errCode;
949 }
950 parcel.ReadUInt32(softwareVersion);
951 parcel.ReadInt(ackCode);
952 parcel.ReadString(schema);
953 errCode = AckPacketDeSerializationTailPart(parcel, packet, softwareVersion);
954 if (parcel.IsError() || errCode != E_OK) {
955 LOGE("[AbilitySync][RequestDeSerialization] DeSerialization failed!");
956 errCode = -E_PARSE_FAIL;
957 goto ERROR_OUT;
958 }
959 packet->SetSoftwareVersion(softwareVersion);
960 packet->SetAckCode(ackCode);
961 packet->SetSchema(schema);
962 errCode = inMsg->SetExternalObject<>(packet);
963 if (errCode == E_OK) {
964 return E_OK;
965 }
966
967 ERROR_OUT:
968 delete packet;
969 return errCode;
970 }
971
SetAbilityRequestBodyInfoInner(uint16_t remoteCommunicatorVersion,AbilitySyncRequestPacket & packet,std::string & schemaStr,uint32_t schemaType) const972 void AbilitySync::SetAbilityRequestBodyInfoInner(uint16_t remoteCommunicatorVersion, AbilitySyncRequestPacket &packet,
973 std::string &schemaStr, uint32_t schemaType) const
974 {
975 // 102 version is forbidden to sync with 103 json-schema or flatbuffer-schema
976 // so schema should put null string while remote is 102 version to avoid this bug.
977 if (remoteCommunicatorVersion == 1) {
978 packet.SetSchema("");
979 packet.SetSchemaType(0);
980 } else {
981 packet.SetSchema(schemaStr);
982 packet.SetSchemaType(schemaType);
983 }
984 }
985
SetAbilityRequestBodyInfo(uint16_t remoteCommunicatorVersion,const ISyncTaskContext * context,AbilitySyncRequestPacket & packet) const986 int AbilitySync::SetAbilityRequestBodyInfo(uint16_t remoteCommunicatorVersion, const ISyncTaskContext *context,
987 AbilitySyncRequestPacket &packet) const
988 {
989 if (storageInterface_ == nullptr) {
990 LOGE("[AbilitySync][FillAbilityRequest] storageInterface is nullptr");
991 return -E_INVALID_ARGS;
992 }
993 uint64_t dbCreateTime;
994 int errCode = (static_cast<SyncGenericInterface *>(storageInterface_))->GetDatabaseCreateTimestamp(dbCreateTime);
995 if (errCode != E_OK) {
996 LOGE("[AbilitySync][FillAbilityRequest] GetDatabaseCreateTimestamp failed, err %d", errCode);
997 return errCode;
998 }
999 SecurityOption option;
1000 GetPacketSecOption(context, option);
1001 std::string schemaStr;
1002 uint32_t schemaType = 0;
1003 if (IsSingleKvVer()) {
1004 SchemaObject schemaObj = (static_cast<SingleVerKvDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1005 schemaStr = schemaObj.ToSchemaString();
1006 schemaType = static_cast<uint32_t>(schemaObj.GetSchemaType());
1007 } else if (IsSingleRelationalVer()) {
1008 auto schemaObj = (static_cast<RelationalDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1009 schemaStr = schemaObj.ToSchemaString();
1010 schemaType = static_cast<uint32_t>(schemaObj.GetSchemaType());
1011 }
1012 DbAbility dbAbility;
1013 errCode = GetDbAbilityInfo(dbAbility);
1014 if (errCode != E_OK) {
1015 LOGE("[AbilitySync][FillAbilityRequest] GetDbAbility failed, err %d", errCode);
1016 return errCode;
1017 }
1018 auto [err, localSchemaVer] = metadata_->GetLocalSchemaVersion();
1019 if (err != E_OK) {
1020 LOGE("[AbilitySync][FillAbilityRequest] GetLocalSchemaVersion failed, err %d", err);
1021 return err;
1022 }
1023 SetAbilityRequestBodyInfoInner(remoteCommunicatorVersion, packet, schemaStr, schemaType);
1024 packet.SetProtocolVersion(ABILITY_SYNC_VERSION_V1);
1025 packet.SetSoftwareVersion(SOFTWARE_VERSION_CURRENT);
1026 packet.SetSecLabel(option.securityLabel);
1027 packet.SetSecFlag(option.securityFlag);
1028 packet.SetDbCreateTime(dbCreateTime);
1029 packet.SetDbAbility(dbAbility);
1030 packet.SetSchemaVersion(localSchemaVer);
1031 LOGI("[AbilitySync][FillRequest] ver=%u,Lab=%d,Flag=%d,dbCreateTime=%" PRId64 ",schemaVer=%" PRId64,
1032 SOFTWARE_VERSION_CURRENT, option.securityLabel, option.securityFlag, dbCreateTime, localSchemaVer);
1033 return E_OK;
1034 }
1035
SetAbilityAckBodyInfo(const ISyncTaskContext * context,int ackCode,bool isAckNotify,AbilitySyncAckPacket & ackPacket) const1036 int AbilitySync::SetAbilityAckBodyInfo(const ISyncTaskContext *context, int ackCode, bool isAckNotify,
1037 AbilitySyncAckPacket &ackPacket) const
1038 {
1039 ackPacket.SetProtocolVersion(ABILITY_SYNC_VERSION_V1);
1040 ackPacket.SetSoftwareVersion(SOFTWARE_VERSION_CURRENT);
1041 if (!isAckNotify) {
1042 SecurityOption option;
1043 GetPacketSecOption(context, option);
1044 ackPacket.SetSecLabel(option.securityLabel);
1045 ackPacket.SetSecFlag(option.securityFlag);
1046 uint64_t dbCreateTime = 0;
1047 int errCode =
1048 (static_cast<SyncGenericInterface *>(storageInterface_))->GetDatabaseCreateTimestamp(dbCreateTime);
1049 if (errCode != E_OK) {
1050 LOGE("[AbilitySync][SyncStart] GetDatabaseCreateTimestamp failed, err %d", errCode);
1051 ackCode = errCode;
1052 }
1053 DbAbility dbAbility;
1054 errCode = GetDbAbilityInfo(dbAbility);
1055 if (errCode != E_OK) {
1056 LOGE("[AbilitySync][FillAbilityRequest] GetDbAbility failed, err %d", errCode);
1057 return errCode;
1058 }
1059 ackPacket.SetDbCreateTime(dbCreateTime);
1060 ackPacket.SetDbAbility(dbAbility);
1061 }
1062 auto [ret, schemaVersion] = metadata_->GetLocalSchemaVersion();
1063 if (ret != E_OK) {
1064 return ret;
1065 }
1066 ackPacket.SetAckCode(ackCode);
1067 ackPacket.SetSchemaVersion(schemaVersion);
1068 return E_OK;
1069 }
1070
SetAbilityAckSchemaInfo(AbilitySyncAckPacket & ackPacket,const ISchema & schemaObj)1071 void AbilitySync::SetAbilityAckSchemaInfo(AbilitySyncAckPacket &ackPacket, const ISchema &schemaObj)
1072 {
1073 ackPacket.SetSchema(schemaObj.ToSchemaString());
1074 ackPacket.SetSchemaType(static_cast<uint32_t>(schemaObj.GetSchemaType()));
1075 }
1076
SetAbilityAckSyncOpinionInfo(AbilitySyncAckPacket & ackPacket,SyncOpinion localOpinion)1077 void AbilitySync::SetAbilityAckSyncOpinionInfo(AbilitySyncAckPacket &ackPacket, SyncOpinion localOpinion)
1078 {
1079 ackPacket.SetPermitSync(localOpinion.permitSync);
1080 ackPacket.SetRequirePeerConvert(localOpinion.requirePeerConvert);
1081 }
1082
GetDbAbilityInfo(DbAbility & dbAbility)1083 int AbilitySync::GetDbAbilityInfo(DbAbility &dbAbility)
1084 {
1085 int errCode = E_OK;
1086 for (const auto &item : SyncConfig::ABILITYBITS) {
1087 errCode = dbAbility.SetAbilityItem(item, SUPPORT_MARK);
1088 if (errCode != E_OK) {
1089 return errCode;
1090 }
1091 }
1092 return errCode;
1093 }
1094
AckMsgCheck(const Message * message,ISyncTaskContext * context) const1095 int AbilitySync::AckMsgCheck(const Message *message, ISyncTaskContext *context) const
1096 {
1097 if (message == nullptr || context == nullptr) {
1098 return -E_INVALID_ARGS;
1099 }
1100 if (message->GetErrorNo() == E_FEEDBACK_UNKNOWN_MESSAGE) {
1101 LOGE("[AbilitySync][AckMsgCheck] Remote device dose not support this message id");
1102 context->SetRemoteSoftwareVersion(SOFTWARE_VERSION_EARLIEST);
1103 context->SetTaskErrCode(-E_FEEDBACK_UNKNOWN_MESSAGE);
1104 return -E_FEEDBACK_UNKNOWN_MESSAGE;
1105 }
1106 if (message->GetErrorNo() == E_FEEDBACK_COMMUNICATOR_NOT_FOUND) {
1107 LOGE("[AbilitySync][AckMsgCheck] Remote db is closed");
1108 context->SetTaskErrCode(-E_FEEDBACK_COMMUNICATOR_NOT_FOUND);
1109 return -E_FEEDBACK_COMMUNICATOR_NOT_FOUND;
1110 }
1111 const AbilitySyncAckPacket *packet = message->GetObject<AbilitySyncAckPacket>();
1112 if (packet == nullptr) {
1113 return -E_INVALID_ARGS;
1114 }
1115 int ackCode = packet->GetAckCode();
1116 if (ackCode != E_OK) {
1117 LOGE("[AbilitySync][AckMsgCheck] received an errCode %d", ackCode);
1118 context->SetTaskErrCode(ackCode);
1119 return ackCode;
1120 }
1121 return E_OK;
1122 }
1123
IsSingleKvVer() const1124 bool AbilitySync::IsSingleKvVer() const
1125 {
1126 return storageInterface_->GetInterfaceType() == ISyncInterface::SYNC_SVD;
1127 }
IsSingleRelationalVer() const1128 bool AbilitySync::IsSingleRelationalVer() const
1129 {
1130 #ifdef RELATIONAL_STORE
1131 return storageInterface_->GetInterfaceType() == ISyncInterface::SYNC_RELATION;
1132 #else
1133 return false;
1134 #endif
1135 }
1136
HandleRequestRecv(const Message * message,ISyncTaskContext * context,bool isCompatible)1137 int AbilitySync::HandleRequestRecv(const Message *message, ISyncTaskContext *context, bool isCompatible)
1138 {
1139 const AbilitySyncRequestPacket *packet = message->GetObject<AbilitySyncRequestPacket>();
1140 if (packet == nullptr) {
1141 return -E_INVALID_ARGS;
1142 }
1143 uint32_t remoteSoftwareVersion = packet->GetSoftwareVersion();
1144 int ackCode;
1145 std::string schema = packet->GetSchema();
1146 if (remoteSoftwareVersion <= SOFTWARE_VERSION_RELEASE_2_0) {
1147 LOGI("[AbilitySync][RequestRecv] remote version = %u, CheckSchemaCompatible = %d",
1148 remoteSoftwareVersion, isCompatible);
1149 return SendAckWithEmptySchema(context, message, E_OK, false);
1150 }
1151 HandleVersionV3RequestParam(packet, context);
1152 if (SecLabelCheck(packet)) {
1153 ackCode = E_OK;
1154 } else {
1155 ackCode = -E_SECURITY_OPTION_CHECK_ERROR;
1156 }
1157 if (ackCode == E_OK && remoteSoftwareVersion > SOFTWARE_VERSION_RELEASE_3_0) {
1158 ackCode = metadata_->SetDbCreateTime(deviceId_, packet->GetDbCreateTime(), true);
1159 }
1160 if (ackCode == E_OK && remoteSoftwareVersion >= SOFTWARE_VERSION_RELEASE_9_0) {
1161 ackCode = metadata_->SetRemoteSchemaVersion(context->GetDeviceId(), packet->GetSchemaVersion());
1162 }
1163 AbilitySyncAckPacket ackPacket;
1164 if (IsSingleRelationalVer()) {
1165 ackPacket.SetRelationalSyncOpinion(MakeRelationSyncOpinion(packet, schema));
1166 } else {
1167 SetAbilityAckSyncOpinionInfo(ackPacket, MakeKvSyncOpinion(packet, schema, context));
1168 }
1169 LOGI("[AbilitySync][RequestRecv] remote dev=%s,ver=%u,schemaCompatible=%d", STR_MASK(deviceId_),
1170 remoteSoftwareVersion, isCompatible);
1171 int errCode = SendAck(context, message, ackCode, false, ackPacket);
1172 return ackCode != E_OK ? ackCode : errCode;
1173 }
1174
SendAck(const ISyncTaskContext * context,const Message * message,int ackCode,bool isAckNotify,AbilitySyncAckPacket & ackPacket)1175 int AbilitySync::SendAck(const ISyncTaskContext *context, const Message *message, int ackCode, bool isAckNotify,
1176 AbilitySyncAckPacket &ackPacket)
1177 {
1178 int errCode = SetAbilityAckBodyInfo(context, ackCode, isAckNotify, ackPacket);
1179 if (errCode != E_OK) {
1180 return errCode;
1181 }
1182 if (IsSingleRelationalVer()) {
1183 auto schemaObj = (static_cast<RelationalDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1184 SetAbilityAckSchemaInfo(ackPacket, schemaObj);
1185 } else if (IsSingleKvVer()) {
1186 SchemaObject schemaObject = static_cast<SingleVerKvDBSyncInterface *>(storageInterface_)->GetSchemaInfo();
1187 SetAbilityAckSchemaInfo(ackPacket, schemaObject);
1188 }
1189 return SendAck(message, ackPacket, isAckNotify);
1190 }
1191
SendAckWithEmptySchema(const ISyncTaskContext * context,const Message * message,int ackCode,bool isAckNotify)1192 int AbilitySync::SendAckWithEmptySchema(const ISyncTaskContext *context, const Message *message, int ackCode,
1193 bool isAckNotify)
1194 {
1195 AbilitySyncAckPacket ackPacket;
1196 int errCode = SetAbilityAckBodyInfo(context, ackCode, isAckNotify, ackPacket);
1197 if (errCode != E_OK) {
1198 return errCode;
1199 }
1200 SetAbilityAckSchemaInfo(ackPacket, SchemaObject());
1201 return SendAck(message, ackPacket, isAckNotify);
1202 }
1203
SendAck(const Message * inMsg,const AbilitySyncAckPacket & ackPacket,bool isAckNotify)1204 int AbilitySync::SendAck(const Message *inMsg, const AbilitySyncAckPacket &ackPacket, bool isAckNotify)
1205 {
1206 Message *ackMessage = new (std::nothrow) Message(ABILITY_SYNC_MESSAGE);
1207 if (ackMessage == nullptr) {
1208 LOGE("[AbilitySync][SendAck] message create failed, may be memleak!");
1209 return -E_OUT_OF_MEMORY;
1210 }
1211 int errCode = ackMessage->SetCopiedObject<>(ackPacket);
1212 if (errCode != E_OK) {
1213 LOGE("[AbilitySync][SendAck] SetCopiedObject failed, err %d", errCode);
1214 delete ackMessage;
1215 ackMessage = nullptr;
1216 return errCode;
1217 }
1218 (!isAckNotify) ? ackMessage->SetMessageType(TYPE_RESPONSE) : ackMessage->SetMessageType(TYPE_NOTIFY);
1219 ackMessage->SetTarget(deviceId_);
1220 ackMessage->SetSessionId(inMsg->GetSessionId());
1221 ackMessage->SetSequenceId(inMsg->GetSequenceId());
1222 SendConfig conf;
1223 SetSendConfigParam(storageInterface_->GetDbProperties(), deviceId_, false, SEND_TIME_OUT, conf);
1224 errCode = communicator_->SendMessage(deviceId_, ackMessage, conf);
1225 if (errCode != E_OK) {
1226 LOGE("[AbilitySync][SendAck] SendPacket failed, err %d", errCode);
1227 delete ackMessage;
1228 ackMessage = nullptr;
1229 }
1230 return errCode;
1231 }
1232
MakeKvSyncOpinion(const AbilitySyncRequestPacket * packet,const std::string & remoteSchema,ISyncTaskContext * context)1233 SyncOpinion AbilitySync::MakeKvSyncOpinion(const AbilitySyncRequestPacket *packet,
1234 const std::string &remoteSchema, ISyncTaskContext *context)
1235 {
1236 uint8_t remoteSchemaType = packet->GetSchemaType();
1237 SchemaObject localSchema = (static_cast<SingleVerKvDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1238 SyncOpinion localSyncOpinion = SchemaNegotiate::MakeLocalSyncOpinion(localSchema, remoteSchema, remoteSchemaType);
1239 if (IsBothKvAndOptAbilitySync(context->GetRemoteSoftwareVersion(),
1240 localSchema.GetSchemaType(), remoteSchemaType)) { // LCOV_EXCL_BR_LINE
1241 // both kv no need convert
1242 SyncStrategy localStrategy;
1243 localStrategy.permitSync = true;
1244 (static_cast<SingleVerKvSyncTaskContext *>(context))->SetSyncStrategy(localStrategy, true);
1245 SetAbilitySyncFinishedStatus(true, *context);
1246 }
1247 return localSyncOpinion;
1248 }
1249
MakeRelationSyncOpinion(const AbilitySyncRequestPacket * packet,const std::string & remoteSchema) const1250 RelationalSyncOpinion AbilitySync::MakeRelationSyncOpinion(const AbilitySyncRequestPacket *packet,
1251 const std::string &remoteSchema) const
1252 {
1253 uint8_t remoteSchemaType = packet->GetSchemaType();
1254 RelationalSchemaObject localSchema = (static_cast<RelationalDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1255 return SchemaNegotiate::MakeLocalSyncOpinion(localSchema, remoteSchema, remoteSchemaType,
1256 packet->GetSoftwareVersion());
1257 }
1258
HandleKvAckSchemaParam(const AbilitySyncAckPacket * recvPacket,ISyncTaskContext * context,AbilitySyncAckPacket & sendPacket,std::pair<bool,bool> & schemaSyncStatus)1259 int AbilitySync::HandleKvAckSchemaParam(const AbilitySyncAckPacket *recvPacket,
1260 ISyncTaskContext *context, AbilitySyncAckPacket &sendPacket, std::pair<bool, bool> &schemaSyncStatus)
1261 {
1262 std::string remoteSchema = recvPacket->GetSchema();
1263 uint8_t remoteSchemaType = recvPacket->GetSchemaType();
1264 bool permitSync = static_cast<bool>(recvPacket->GetPermitSync());
1265 bool requirePeerConvert = static_cast<bool>(recvPacket->GetRequirePeerConvert());
1266 SyncOpinion remoteOpinion = {permitSync, requirePeerConvert, true};
1267 SchemaObject localSchema = (static_cast<SingleVerKvDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1268 SyncOpinion syncOpinion = SchemaNegotiate::MakeLocalSyncOpinion(localSchema, remoteSchema, remoteSchemaType);
1269 SyncStrategy localStrategy = SchemaNegotiate::ConcludeSyncStrategy(syncOpinion, remoteOpinion);
1270 SetAbilityAckSyncOpinionInfo(sendPacket, syncOpinion);
1271 (static_cast<SingleVerKvSyncTaskContext *>(context))->SetSyncStrategy(localStrategy, true);
1272 schemaSyncStatus = {
1273 localStrategy.permitSync,
1274 true
1275 };
1276 if (localStrategy.permitSync) { // LCOV_EXCL_BR_LINE
1277 RecordAbilitySyncFinish(recvPacket->GetSchemaVersion(), *context);
1278 }
1279 if (IsBothKvAndOptAbilitySync(context->GetRemoteSoftwareVersion(),
1280 localSchema.GetSchemaType(), remoteSchemaType)) { // LCOV_EXCL_BR_LINE
1281 return -E_ABILITY_SYNC_FINISHED;
1282 }
1283 return E_OK;
1284 }
1285
HandleRelationAckSchemaParam(const AbilitySyncAckPacket * recvPacket,AbilitySyncAckPacket & sendPacket,ISyncTaskContext * context,bool sendOpinion,std::pair<bool,bool> & schemaSyncStatus)1286 int AbilitySync::HandleRelationAckSchemaParam(const AbilitySyncAckPacket *recvPacket, AbilitySyncAckPacket &sendPacket,
1287 ISyncTaskContext *context, bool sendOpinion, std::pair<bool, bool> &schemaSyncStatus)
1288 {
1289 std::string remoteSchema = recvPacket->GetSchema();
1290 uint8_t remoteSchemaType = recvPacket->GetSchemaType();
1291 auto localSchema = (static_cast<RelationalDBSyncInterface *>(storageInterface_))->GetSchemaInfo();
1292 auto localOpinion = SchemaNegotiate::MakeLocalSyncOpinion(localSchema, remoteSchema, remoteSchemaType,
1293 recvPacket->GetSoftwareVersion());
1294 auto localStrategy = SchemaNegotiate::ConcludeSyncStrategy(localOpinion,
1295 recvPacket->GetRelationalSyncOpinion());
1296 (static_cast<SingleVerRelationalSyncTaskContext *>(context))->SetRelationalSyncStrategy(localStrategy, true);
1297 bool permitSync = std::any_of(localStrategy.begin(), localStrategy.end(),
1298 [] (const std::pair<std::string, SyncStrategy> &it) {
1299 return it.second.permitSync;
1300 });
1301 if (permitSync) {
1302 int innerErrCode = (static_cast<RelationalDBSyncInterface *>(storageInterface_)->SaveRemoteDeviceSchema(
1303 deviceId_, remoteSchema, remoteSchemaType));
1304 if (innerErrCode != E_OK) {
1305 LOGE("[AbilitySync][AckRecv] save remote device Schema failed,errCode=%d", innerErrCode);
1306 return innerErrCode;
1307 }
1308 }
1309 int errCode = (static_cast<RelationalDBSyncInterface *>(storageInterface_))->
1310 CreateDistributedDeviceTable(context->GetDeviceId(), localStrategy);
1311 if (errCode != E_OK) {
1312 LOGE("[AbilitySync][AckRecv] create distributed device table failed,errCode=%d", errCode);
1313 }
1314 if (sendOpinion) {
1315 sendPacket.SetRelationalSyncOpinion(localOpinion);
1316 }
1317 auto singleVerContext = static_cast<SingleVerSyncTaskContext *>(context);
1318 auto strategy = localStrategy.find(singleVerContext->GetQuery().GetRelationTableName());
1319 schemaSyncStatus = {
1320 !(strategy == localStrategy.end()) && strategy->second.permitSync,
1321 true
1322 };
1323 if (permitSync) {
1324 RecordAbilitySyncFinish(recvPacket->GetSchemaVersion(), *context);
1325 }
1326 return errCode;
1327 }
1328
AckRecvWithHighVersion(const Message * message,ISyncTaskContext * context,const AbilitySyncAckPacket * packet)1329 int AbilitySync::AckRecvWithHighVersion(const Message *message, ISyncTaskContext *context,
1330 const AbilitySyncAckPacket *packet)
1331 {
1332 HandleVersionV3AckSecOptionParam(packet, context);
1333 AbilitySyncAckPacket ackPacket;
1334 std::pair<bool, bool> schemaSyncStatus;
1335 int errCode = E_OK;
1336 if (context->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_3_0) {
1337 errCode = metadata_->SetDbCreateTime(deviceId_, packet->GetDbCreateTime(), true);
1338 if (errCode != E_OK) {
1339 LOGE("[AbilitySync][AckRecv] set db create time failed,errCode=%d", errCode);
1340 context->SetTaskErrCode(errCode);
1341 return errCode;
1342 }
1343 }
1344 errCode = HandleVersionV3AckSchemaParam(packet, ackPacket, context, true, schemaSyncStatus);
1345 DbAbility remoteDbAbility = packet->GetDbAbility();
1346 auto singleVerContext = static_cast<SingleVerSyncTaskContext *>(context);
1347 singleVerContext->SetDbAbility(remoteDbAbility);
1348 if (errCode == -E_ABILITY_SYNC_FINISHED) {
1349 return errCode;
1350 }
1351 if (errCode != E_OK) {
1352 context->SetTaskErrCode(errCode);
1353 return errCode;
1354 }
1355 if (!schemaSyncStatus.first) {
1356 singleVerContext->SetTaskErrCode(-E_SCHEMA_MISMATCH);
1357 LOGE("[AbilitySync][AckRecv] scheme check failed");
1358 return -E_SCHEMA_MISMATCH;
1359 }
1360 (void)SendAck(context, message, AbilitySync::CHECK_SUCCESS, true, ackPacket);
1361 return E_OK;
1362 }
1363
TransformSecLabelIfNeed(int32_t originLabel,int targetLabel)1364 int32_t AbilitySync::TransformSecLabelIfNeed(int32_t originLabel, int targetLabel)
1365 {
1366 if ((originLabel == S0 && targetLabel == S1) || (originLabel == S1 && targetLabel == S0)) {
1367 LOGI("[AbilitySync] Accept SecLabel From %d To %d", originLabel, targetLabel);
1368 return targetLabel;
1369 }
1370 return originLabel;
1371 }
1372
IsBothKvAndOptAbilitySync(uint32_t remoteVersion,SchemaType localType,uint8_t remoteType)1373 bool AbilitySync::IsBothKvAndOptAbilitySync(uint32_t remoteVersion, SchemaType localType, uint8_t remoteType)
1374 {
1375 return remoteVersion >= SOFTWARE_VERSION_RELEASE_8_0 && localType == SchemaType::NONE &&
1376 static_cast<SchemaType>(remoteType) == SchemaType::NONE;
1377 }
1378
InitAbilitySyncFinishStatus(ISyncTaskContext & context)1379 void AbilitySync::InitAbilitySyncFinishStatus(ISyncTaskContext &context)
1380 {
1381 if (!metadata_->IsAbilitySyncFinish(context.GetDeviceId())) {
1382 return;
1383 }
1384 LOGI("[AbilitySync] Mark ability sync finish from db status");
1385 syncFinished_ = true;
1386 if (context.GetRemoteSoftwareVersion() == 0u) { // LCOV_EXCL_BR_LINE
1387 LOGI("[AbilitySync] Init remote version with default");
1388 context.SetRemoteSoftwareVersion(SOFTWARE_VERSION_RELEASE_9_0); // remote version >= 109
1389 }
1390 InitRemoteDBAbility(context);
1391 }
1392
InitRemoteDBAbility(ISyncTaskContext & context)1393 void AbilitySync::InitRemoteDBAbility(ISyncTaskContext &context)
1394 {
1395 DbAbility ability;
1396 int errCode = GetDbAbilityInfo(ability);
1397 if (errCode != E_OK) {
1398 return;
1399 }
1400 context.SetDbAbility(ability);
1401 auto version = static_cast<uint32_t>(metadata_->GetRemoteSoftwareVersion(context.GetDeviceId()));
1402 if (version > 0) {
1403 context.SetRemoteSoftwareVersion(version);
1404 }
1405 }
1406
RecordAbilitySyncFinish(uint64_t remoteSchemaVersion,ISyncTaskContext & context)1407 void AbilitySync::RecordAbilitySyncFinish(uint64_t remoteSchemaVersion, ISyncTaskContext &context)
1408 {
1409 SetAbilitySyncFinishedStatus(true, context);
1410 if (context.GetRemoteSoftwareVersion() >= SOFTWARE_VERSION_RELEASE_9_0) { // LCOV_EXCL_BR_LINE
1411 (void)metadata_->SetRemoteSchemaVersion(deviceId_, remoteSchemaVersion);
1412 }
1413 }
1414 } // namespace DistributedDB