简柏林 / thingskit · Commits

Commit 6cdb508c71041fb15757ef2e7d9182e48ea18601

Authored by Igor Kulikov
Committed by GitHub
1 parent fcb0c041

Claiming devices implementation

Inline Side-by-side
Showing 30 changed files with 743 additions and 109 deletions
... ... @@ -28,7 +28,16 @@ import org.springframework.security.core.Authentication;
28 28 import org.springframework.security.core.context.SecurityContextHolder;
29 29 import org.springframework.web.bind.annotation.ExceptionHandler;
30 30 import org.thingsboard.server.actors.service.ActorService;
31   -import org.thingsboard.server.common.data.*;
  31 +import org.thingsboard.server.common.data.Customer;
  32 +import org.thingsboard.server.common.data.Dashboard;
  33 +import org.thingsboard.server.common.data.DashboardInfo;
  34 +import org.thingsboard.server.common.data.DataConstants;
  35 +import org.thingsboard.server.common.data.Device;
  36 +import org.thingsboard.server.common.data.EntityType;
  37 +import org.thingsboard.server.common.data.EntityView;
  38 +import org.thingsboard.server.common.data.HasName;
  39 +import org.thingsboard.server.common.data.Tenant;
  40 +import org.thingsboard.server.common.data.User;
32 41 import org.thingsboard.server.common.data.alarm.Alarm;
33 42 import org.thingsboard.server.common.data.alarm.AlarmId;
34 43 import org.thingsboard.server.common.data.alarm.AlarmInfo;
... ... @@ -36,7 +45,19 @@ import org.thingsboard.server.common.data.asset.Asset;
36 45 import org.thingsboard.server.common.data.audit.ActionType;
37 46 import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
38 47 import org.thingsboard.server.common.data.exception.ThingsboardException;
39   -import org.thingsboard.server.common.data.id.*;
  48 +import org.thingsboard.server.common.data.id.AssetId;
  49 +import org.thingsboard.server.common.data.id.CustomerId;
  50 +import org.thingsboard.server.common.data.id.DashboardId;
  51 +import org.thingsboard.server.common.data.id.DeviceId;
  52 +import org.thingsboard.server.common.data.id.EntityId;
  53 +import org.thingsboard.server.common.data.id.EntityIdFactory;
  54 +import org.thingsboard.server.common.data.id.EntityViewId;
  55 +import org.thingsboard.server.common.data.id.RuleChainId;
  56 +import org.thingsboard.server.common.data.id.RuleNodeId;
  57 +import org.thingsboard.server.common.data.id.TenantId;
  58 +import org.thingsboard.server.common.data.id.UserId;
  59 +import org.thingsboard.server.common.data.id.WidgetTypeId;
  60 +import org.thingsboard.server.common.data.id.WidgetsBundleId;
40 61 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
41 62 import org.thingsboard.server.common.data.kv.DataType;
42 63 import org.thingsboard.server.common.data.page.TextPageLink;
... ... @@ -45,7 +66,6 @@ import org.thingsboard.server.common.data.plugin.ComponentDescriptor;
45 66 import org.thingsboard.server.common.data.plugin.ComponentType;
46 67 import org.thingsboard.server.common.data.rule.RuleChain;
47 68 import org.thingsboard.server.common.data.rule.RuleNode;
48   -import org.thingsboard.server.common.data.security.Authority;
49 69 import org.thingsboard.server.common.data.widget.WidgetType;
50 70 import org.thingsboard.server.common.data.widget.WidgetsBundle;
51 71 import org.thingsboard.server.common.msg.TbMsg;
... ... @@ -53,13 +73,13 @@ import org.thingsboard.server.common.msg.TbMsgDataType;
53 73 import org.thingsboard.server.common.msg.TbMsgMetaData;
54 74 import org.thingsboard.server.common.msg.cluster.SendToClusterMsg;
55 75 import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg;
56   -import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
57 76 import org.thingsboard.server.dao.alarm.AlarmService;
58 77 import org.thingsboard.server.dao.asset.AssetService;
59 78 import org.thingsboard.server.dao.attributes.AttributesService;
60 79 import org.thingsboard.server.dao.audit.AuditLogService;
61 80 import org.thingsboard.server.dao.customer.CustomerService;
62 81 import org.thingsboard.server.dao.dashboard.DashboardService;
  82 +import org.thingsboard.server.dao.device.ClaimDevicesService;
63 83 import org.thingsboard.server.dao.device.DeviceCredentialsService;
64 84 import org.thingsboard.server.dao.device.DeviceService;
65 85 import org.thingsboard.server.dao.entityview.EntityViewService;
... ... @@ -162,6 +182,9 @@ public abstract class BaseController {
162 182 @Autowired
163 183 protected AttributesService attributesService;
164 184
  185 + @Autowired
  186 + protected ClaimDevicesService claimDevicesService;
  187 +
165 188 @Value("${server.log_controller_error_stack_trace}")
166 189 @Getter
167 190 private boolean logControllerErrorStackTrace;
... ...
... ... @@ -15,8 +15,11 @@
15 15 */
16 16 package org.thingsboard.server.controller;
17 17
  18 +import com.google.common.util.concurrent.FutureCallback;
  19 +import com.google.common.util.concurrent.Futures;
18 20 import com.google.common.util.concurrent.ListenableFuture;
19 21 import org.springframework.http.HttpStatus;
  22 +import org.springframework.http.ResponseEntity;
20 23 import org.springframework.security.access.prepost.PreAuthorize;
21 24 import org.springframework.web.bind.annotation.PathVariable;
22 25 import org.springframework.web.bind.annotation.RequestBody;
... ... @@ -26,27 +29,31 @@ import org.springframework.web.bind.annotation.RequestParam;
26 29 import org.springframework.web.bind.annotation.ResponseBody;
27 30 import org.springframework.web.bind.annotation.ResponseStatus;
28 31 import org.springframework.web.bind.annotation.RestController;
  32 +import org.springframework.web.context.request.async.DeferredResult;
29 33 import org.thingsboard.server.common.data.Customer;
  34 +import org.thingsboard.server.common.data.DataConstants;
30 35 import org.thingsboard.server.common.data.Device;
31 36 import org.thingsboard.server.common.data.EntitySubtype;
32 37 import org.thingsboard.server.common.data.EntityType;
33 38 import org.thingsboard.server.common.data.audit.ActionType;
34 39 import org.thingsboard.server.common.data.device.DeviceSearchQuery;
35   -import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
36 40 import org.thingsboard.server.common.data.exception.ThingsboardException;
37 41 import org.thingsboard.server.common.data.id.CustomerId;
38 42 import org.thingsboard.server.common.data.id.DeviceId;
39 43 import org.thingsboard.server.common.data.id.TenantId;
40 44 import org.thingsboard.server.common.data.page.TextPageData;
41 45 import org.thingsboard.server.common.data.page.TextPageLink;
42   -import org.thingsboard.server.common.data.security.Authority;
43 46 import org.thingsboard.server.common.data.security.DeviceCredentials;
  47 +import org.thingsboard.server.controller.claim.data.ClaimRequest;
  48 +import org.thingsboard.server.dao.device.claim.ClaimResponse;
44 49 import org.thingsboard.server.dao.exception.IncorrectParameterException;
45 50 import org.thingsboard.server.dao.model.ModelConstants;
46 51 import org.thingsboard.server.service.security.model.SecurityUser;
47 52 import org.thingsboard.server.service.security.permission.Operation;
48 53 import org.thingsboard.server.service.security.permission.Resource;
49 54
  55 +import javax.annotation.Nullable;
  56 +import java.io.IOException;
50 57 import java.util.ArrayList;
51 58 import java.util.List;
52 59 import java.util.stream.Collectors;
... ... @@ -55,7 +62,8 @@ import java.util.stream.Collectors;
55 62 @RequestMapping("/api")
56 63 public class DeviceController extends BaseController {
57 64
58   - public static final String DEVICE_ID = "deviceId";
  65 + private static final String DEVICE_ID = "deviceId";
  66 + private static final String DEVICE_NAME = "deviceName";
59 67
60 68 @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
61 69 @RequestMapping(value = "/device/{deviceId}", method = RequestMethod.GET)
... ... @@ -379,4 +387,91 @@ public class DeviceController extends BaseController {
379 387 throw handleException(e);
380 388 }
381 389 }
  390 +
  391 + @PreAuthorize("hasAnyAuthority('CUSTOMER_USER')")
  392 + @RequestMapping(value = "/customer/device/{deviceName}/claim", method = RequestMethod.POST)
  393 + @ResponseBody
  394 + public DeferredResult<ResponseEntity> claimDevice(@PathVariable(DEVICE_NAME) String deviceName,
  395 + @RequestBody(required = false) ClaimRequest claimRequest) throws ThingsboardException {
  396 + checkParameter(DEVICE_NAME, deviceName);
  397 + try {
  398 + final DeferredResult<ResponseEntity> deferredResult = new DeferredResult<>();
  399 +
  400 + SecurityUser user = getCurrentUser();
  401 + TenantId tenantId = user.getTenantId();
  402 + CustomerId customerId = user.getCustomerId();
  403 +
  404 + Device device = checkNotNull(deviceService.findDeviceByTenantIdAndName(tenantId, deviceName));
  405 + accessControlService.checkPermission(user, Resource.DEVICE, Operation.CLAIM_DEVICES,
  406 + device.getId(), device);
  407 + String secretKey = getSecretKey(claimRequest);
  408 +
  409 + ListenableFuture<ClaimResponse> future = claimDevicesService.claimDevice(device, customerId, secretKey);
  410 + Futures.addCallback(future, new FutureCallback<ClaimResponse>() {
  411 + @Override
  412 + public void onSuccess(@Nullable ClaimResponse result) {
  413 + HttpStatus status;
  414 + if (result.equals(ClaimResponse.SUCCESS)) {
  415 + status = HttpStatus.OK;
  416 + } else {
  417 + status = HttpStatus.BAD_REQUEST;
  418 + }
  419 + deferredResult.setResult(new ResponseEntity<>(result, status));
  420 + }
  421 +
  422 + @Override
  423 + public void onFailure(Throwable t) {
  424 + deferredResult.setErrorResult(t);
  425 + }
  426 + });
  427 + return deferredResult;
  428 + } catch (Exception e) {
  429 + throw handleException(e);
  430 + }
  431 + }
  432 +
  433 + @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
  434 + @RequestMapping(value = "/customer/device/{deviceName}/claim", method = RequestMethod.DELETE)
  435 + @ResponseStatus(value = HttpStatus.OK)
  436 + public DeferredResult<ResponseEntity> reClaimDevice(@PathVariable(DEVICE_NAME) String deviceName) throws ThingsboardException {
  437 + checkParameter(DEVICE_NAME, deviceName);
  438 + try {
  439 + final DeferredResult<ResponseEntity> deferredResult = new DeferredResult<>();
  440 +
  441 + SecurityUser user = getCurrentUser();
  442 + TenantId tenantId = user.getTenantId();
  443 +
  444 + Device device = checkNotNull(deviceService.findDeviceByTenantIdAndName(tenantId, deviceName));
  445 + accessControlService.checkPermission(user, Resource.DEVICE, Operation.CLAIM_DEVICES,
  446 + device.getId(), device);
  447 +
  448 + ListenableFuture<List<Void>> future = claimDevicesService.reClaimDevice(tenantId, device);
  449 + Futures.addCallback(future, new FutureCallback<List<Void>>() {
  450 + @Override
  451 + public void onSuccess(@Nullable List<Void> result) {
  452 + if (result != null) {
  453 + deferredResult.setResult(new ResponseEntity(HttpStatus.OK));
  454 + } else {
  455 + deferredResult.setResult(new ResponseEntity(HttpStatus.BAD_REQUEST));
  456 + }
  457 + }
  458 +
  459 + @Override
  460 + public void onFailure(Throwable t) {
  461 + deferredResult.setErrorResult(t);
  462 + }
  463 + });
  464 + return deferredResult;
  465 + } catch (Exception e) {
  466 + throw handleException(e);
  467 + }
  468 + }
  469 +
  470 + private String getSecretKey(ClaimRequest claimRequest) throws IOException {
  471 + String secretKey = claimRequest.getSecretKey();
  472 + if (secretKey != null) {
  473 + return secretKey;
  474 + }
  475 + return DataConstants.DEFAULT_SECRET_KEY;
  476 + }
382 477 }
... ...
  1 +/**
  2 + * Copyright © 2016-2019 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.controller.claim.data;
  17 +
  18 +import lombok.Data;
  19 +
  20 +@Data
  21 +public class ClaimRequest {
  22 +
  23 + private final String secretKey;
  24 +
  25 +}
... ...
application/src/main/java/org/thingsboard/server/service/security/permission/CustomerUserPermissions.java renamed from application/src/main/java/org/thingsboard/server/service/security/permission/CustomerUserPremissions.java
... ... @@ -16,20 +16,20 @@
16 16 package org.thingsboard.server.service.security.permission;
17 17
18 18 import org.springframework.stereotype.Component;
19   -import org.thingsboard.server.common.data.*;
  19 +import org.thingsboard.server.common.data.DashboardInfo;
  20 +import org.thingsboard.server.common.data.HasCustomerId;
  21 +import org.thingsboard.server.common.data.HasTenantId;
  22 +import org.thingsboard.server.common.data.User;
20 23 import org.thingsboard.server.common.data.id.DashboardId;
21 24 import org.thingsboard.server.common.data.id.EntityId;
22   -import org.thingsboard.server.common.data.id.TenantId;
23 25 import org.thingsboard.server.common.data.id.UserId;
24 26 import org.thingsboard.server.common.data.security.Authority;
25 27 import org.thingsboard.server.service.security.model.SecurityUser;
26 28
27   -import java.util.HashMap;
  29 +@Component(value = "customerUserPermissions")
  30 +public class CustomerUserPermissions extends AbstractPermissions {
28 31
29   -@Component(value="customerUserPermissions")
30   -public class CustomerUserPremissions extends AbstractPermissions {
31   -
32   - public CustomerUserPremissions() {
  32 + public CustomerUserPermissions() {
33 33 super();
34 34 put(Resource.ALARM, TenantAdminPermissions.tenantEntityPermissionChecker);
35 35 put(Resource.ASSET, customerEntityPermissionChecker);
... ... @@ -44,26 +44,26 @@ public class CustomerUserPremissions extends AbstractPermissions {
44 44
45 45 private static final PermissionChecker customerEntityPermissionChecker =
46 46 new PermissionChecker.GenericPermissionChecker(Operation.READ, Operation.READ_CREDENTIALS,
47   - Operation.READ_ATTRIBUTES, Operation.READ_TELEMETRY, Operation.RPC_CALL) {
  47 + Operation.READ_ATTRIBUTES, Operation.READ_TELEMETRY, Operation.RPC_CALL, Operation.CLAIM_DEVICES) {
48 48
49   - @Override
50   - public boolean hasPermission(SecurityUser user, Operation operation, EntityId entityId, HasTenantId entity) {
  49 + @Override
  50 + public boolean hasPermission(SecurityUser user, Operation operation, EntityId entityId, HasTenantId entity) {
51 51
52   - if (!super.hasPermission(user, operation, entityId, entity)) {
53   - return false;
54   - }
55   - if (!user.getTenantId().equals(entity.getTenantId())) {
56   - return false;
57   - }
58   - if (!(entity instanceof HasCustomerId)) {
59   - return false;
60   - }
61   - if (!user.getCustomerId().equals(((HasCustomerId)entity).getCustomerId())) {
62   - return false;
63   - }
64   - return true;
65   - }
66   - };
  52 + if (!super.hasPermission(user, operation, entityId, entity)) {
  53 + return false;
  54 + }
  55 + if (!user.getTenantId().equals(entity.getTenantId())) {
  56 + return false;
  57 + }
  58 + if (!(entity instanceof HasCustomerId)) {
  59 + return false;
  60 + }
  61 + if (!operation.equals(Operation.CLAIM_DEVICES) && !user.getCustomerId().equals(((HasCustomerId) entity).getCustomerId())) {
  62 + return false;
  63 + }
  64 + return true;
  65 + }
  66 + };
67 67
68 68 private static final PermissionChecker customerPermissionChecker =
69 69 new PermissionChecker.GenericPermissionChecker(Operation.READ, Operation.READ_ATTRIBUTES, Operation.READ_TELEMETRY) {
... ...
... ... @@ -18,6 +18,6 @@ package org.thingsboard.server.service.security.permission;
18 18 public enum Operation {
19 19
20 20 ALL, CREATE, READ, WRITE, DELETE, ASSIGN_TO_CUSTOMER, UNASSIGN_FROM_CUSTOMER, RPC_CALL,
21   - READ_CREDENTIALS, WRITE_CREDENTIALS, READ_ATTRIBUTES, WRITE_ATTRIBUTES, READ_TELEMETRY, WRITE_TELEMETRY
  21 + READ_CREDENTIALS, WRITE_CREDENTIALS, READ_ATTRIBUTES, WRITE_ATTRIBUTES, READ_TELEMETRY, WRITE_TELEMETRY, CLAIM_DEVICES
22 22
23 23 }
... ...
... ... @@ -22,11 +22,31 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
22 22 import org.springframework.stereotype.Service;
23 23 import org.thingsboard.rule.engine.api.util.DonAsynchron;
24 24 import org.thingsboard.server.actors.ActorSystemContext;
  25 +import org.thingsboard.server.common.data.id.DeviceId;
  26 +import org.thingsboard.server.common.data.id.TenantId;
25 27 import org.thingsboard.server.common.msg.cluster.ServerAddress;
26 28 import org.thingsboard.server.common.transport.TransportServiceCallback;
27 29 import org.thingsboard.server.common.transport.service.AbstractTransportService;
  30 +import org.thingsboard.server.dao.device.ClaimDevicesService;
28 31 import org.thingsboard.server.gen.transport.TransportProtos;
29   -import org.thingsboard.server.gen.transport.TransportProtos.*;
  32 +import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
  33 +import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTransportMsg;
  34 +import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
  35 +import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
  36 +import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg;
  37 +import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
  38 +import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
  39 +import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg;
  40 +import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
  41 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
  42 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
  43 +import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
  44 +import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
  45 +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
  46 +import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg;
  47 +import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg;
  48 +import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
  49 +import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
30 50 import org.thingsboard.server.service.cluster.routing.ClusterRoutingService;
31 51 import org.thingsboard.server.service.cluster.rpc.ClusterRpcService;
32 52 import org.thingsboard.server.service.encoding.DataDecodingEncodingService;
... ... @@ -35,6 +55,7 @@ import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWra
35 55 import javax.annotation.PostConstruct;
36 56 import javax.annotation.PreDestroy;
37 57 import java.util.Optional;
  58 +import java.util.UUID;
38 59 import java.util.function.Consumer;
39 60
40 61 /**
... ... @@ -58,6 +79,8 @@ public class LocalTransportService extends AbstractTransportService implements R
58 79 private ClusterRpcService rpcService;
59 80 @Autowired
60 81 private DataDecodingEncodingService encodingService;
  82 + @Autowired
  83 + private ClaimDevicesService claimDevicesService;
61 84
62 85 @PostConstruct
63 86 public void init() {
... ... @@ -151,6 +174,23 @@ public class LocalTransportService extends AbstractTransportService implements R
151 174 }
152 175
153 176 @Override
  177 + protected void registerClaimingInfo(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback<Void> callback) {
  178 + TransportToDeviceActorMsg toDeviceActorMsg = TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setClaimDevice(msg).build();
  179 +
  180 + TransportToDeviceActorMsgWrapper wrapper = new TransportToDeviceActorMsgWrapper(toDeviceActorMsg);
  181 + Optional<ServerAddress> address = routingService.resolveById(wrapper.getDeviceId());
  182 + if (address.isPresent()) {
  183 + rpcService.tell(encodingService.convertToProtoDataMessage(address.get(), wrapper));
  184 + callback.onSuccess(null);
  185 + } else {
  186 + TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB()));
  187 + DeviceId deviceId = new DeviceId(new UUID(msg.getDeviceIdMSB(), msg.getDeviceIdLSB()));
  188 + DonAsynchron.withCallback(claimDevicesService.registerClaimingInfo(tenantId, deviceId, msg.getSecretKey(), msg.getDurationMs()),
  189 + callback::onSuccess, callback::onError);
  190 + }
  191 + }
  192 +
  193 + @Override
154 194 public void process(String nodeId, DeviceActorToTransportMsg msg) {
155 195 process(nodeId, msg, null, null);
156 196 }
... ...
... ... @@ -103,6 +103,11 @@ security:
103 103 user_token_access_enabled: "${SECURITY_USER_TOKEN_ACCESS_ENABLED:true}"
104 104 # Enable/disable case-sensitive username login
105 105 user_login_case_sensitive: "${SECURITY_USER_LOGIN_CASE_SENSITIVE:true}"
  106 + claim:
  107 + # Enable/disable claiming devices, if false -> the device's [claimingAllowed] SERVER_SCOPE attribute must be set to [true] to allow claiming specific device
  108 + allowClaimingByDefault: "${SECURITY_CLAIM_ALLOW_CLAIMING_BY_DEFAULT:true}"
  109 + # Time allowed to claim the device in milliseconds
  110 + duration: "${SECURITY_CLAIM_DURATION:60000}" # 1 minute, note this value must equal claimDevices.timeToLiveInMinutes value
106 111
107 112 # Dashboard parameters
108 113 dashboard:
... ... @@ -261,6 +266,9 @@ caffeine:
261 266 entityViews:
262 267 timeToLiveInMinutes: 1440
263 268 maxSize: 100000
  269 + claimDevices:
  270 + timeToLiveInMinutes: 1
  271 + maxSize: 100000
264 272
265 273 redis:
266 274 # standalone or cluster
... ...
... ... @@ -22,4 +22,5 @@ public class CacheConstants {
22 22 public static final String SESSIONS_CACHE = "sessions";
23 23 public static final String ASSET_CACHE = "assets";
24 24 public static final String ENTITY_VIEW_CACHE = "entityViews";
  25 + public static final String CLAIM_DEVICES_CACHE = "claimDevices";
25 26 }
... ...
... ... @@ -59,4 +59,8 @@ public class DataConstants {
59 59
60 60 public static final String RPC_CALL_FROM_SERVER_TO_DEVICE = "RPC_CALL_FROM_SERVER_TO_DEVICE";
61 61
  62 + public static final String DEFAULT_SECRET_KEY = "";
  63 + public static final String SECRET_KEY_FIELD_NAME = "secretKey";
  64 + public static final String DURATION_MS_FIELD_NAME = "durationMs";
  65 +
62 66 }
... ...
... ... @@ -16,5 +16,5 @@
16 16 package org.thingsboard.server.common.msg.session;
17 17
18 18 public enum FeatureType {
19   - ATTRIBUTES, TELEMETRY, RPC
  19 + ATTRIBUTES, TELEMETRY, RPC, CLAIM
20 20 }
... ...
... ... @@ -28,7 +28,9 @@ public enum SessionMsgType {
28 28
29 29 RULE_ENGINE_ERROR,
30 30
31   - SESSION_OPEN, SESSION_CLOSE;
  31 + SESSION_OPEN, SESSION_CLOSE,
  32 +
  33 + CLAIM_REQUEST();
32 34
33 35 private final boolean requiresRulesProcessing;
34 36
... ...
... ... @@ -19,7 +19,6 @@ import lombok.extern.slf4j.Slf4j;
19 19 import org.eclipse.californium.core.CoapResource;
20 20 import org.eclipse.californium.core.coap.CoAP.ResponseCode;
21 21 import org.eclipse.californium.core.coap.Request;
22   -import org.eclipse.californium.core.coap.Response;
23 22 import org.eclipse.californium.core.network.Exchange;
24 23 import org.eclipse.californium.core.network.ExchangeObserver;
25 24 import org.eclipse.californium.core.server.resources.CoapExchange;
... ... @@ -122,6 +121,9 @@ public class CoapTransportResource extends CoapResource {
122 121 processRequest(exchange, SessionMsgType.TO_SERVER_RPC_REQUEST);
123 122 }
124 123 break;
  124 + case CLAIM:
  125 + processRequest(exchange, SessionMsgType.CLAIM_REQUEST);
  126 + break;
125 127 }
126 128 }
127 129 }
... ... @@ -153,6 +155,11 @@ public class CoapTransportResource extends CoapResource {
153 155 transportContext.getAdaptor().convertToPostTelemetry(sessionId, request),
154 156 new CoapOkCallback(exchange));
155 157 break;
  158 + case CLAIM_REQUEST:
  159 + transportService.process(sessionInfo,
  160 + transportContext.getAdaptor().convertToClaimDevice(sessionId, request, sessionInfo),
  161 + new CoapOkCallback(exchange));
  162 + break;
156 163 case SUBSCRIBE_ATTRIBUTES_REQUEST:
157 164 advanced.setObserver(new CoapExchangeObserverProxy((ExchangeObserver) observerField.get(advanced),
158 165 registerAsyncCoapSession(exchange, request, sessionInfo, sessionId)));
... ... @@ -319,7 +326,7 @@ public class CoapTransportResource extends CoapResource {
319 326
320 327 @Override
321 328 public void onSuccess(Void msg) {
322   - exchange.respond(ResponseCode.VALID);
  329 + exchange.respond(ResponseCode.VALID);
323 330 }
324 331
325 332 @Override
... ...
... ... @@ -22,7 +22,6 @@ import org.thingsboard.server.gen.transport.TransportProtos;
22 22 import org.thingsboard.server.transport.coap.CoapTransportResource;
23 23
24 24 import java.util.UUID;
25   -import java.util.Optional;
26 25
27 26 public interface CoapTransportAdaptor {
28 27
... ... @@ -36,6 +35,8 @@ public interface CoapTransportAdaptor {
36 35
37 36 TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(UUID sessionId, Request inbound) throws AdaptorException;
38 37
  38 + TransportProtos.ClaimDeviceMsg convertToClaimDevice(UUID sessionId, Request inbound, TransportProtos.SessionInfoProto sessionInfo) throws AdaptorException;
  39 +
39 40 Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException;
40 41
41 42 Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException;
... ...
... ... @@ -15,33 +15,36 @@
15 15 */
16 16 package org.thingsboard.server.transport.coap.adaptors;
17 17
18   -import java.util.*;
19   -
20 18 import com.google.gson.JsonElement;
21 19 import com.google.gson.JsonObject;
  20 +import com.google.gson.JsonParser;
  21 +import com.google.gson.JsonSyntaxException;
22 22 import lombok.extern.slf4j.Slf4j;
23 23 import org.eclipse.californium.core.coap.CoAP;
24 24 import org.eclipse.californium.core.coap.Request;
25 25 import org.eclipse.californium.core.coap.Response;
  26 +import org.springframework.stereotype.Component;
26 27 import org.springframework.util.StringUtils;
27   -import org.thingsboard.server.common.msg.kv.AttributesKVMsg;
28   -import org.thingsboard.server.common.msg.session.SessionContext;
  28 +import org.thingsboard.server.common.data.id.DeviceId;
29 29 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
30 30 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
31   -import org.springframework.stereotype.Component;
32   -
33   -import com.google.gson.JsonParser;
34   -import com.google.gson.JsonSyntaxException;
35 31 import org.thingsboard.server.gen.transport.TransportProtos;
36 32 import org.thingsboard.server.transport.coap.CoapTransportResource;
37 33
  34 +import java.util.Arrays;
  35 +import java.util.HashSet;
  36 +import java.util.List;
  37 +import java.util.Optional;
  38 +import java.util.Set;
  39 +import java.util.UUID;
  40 +
38 41 @Component("JsonCoapAdaptor")
39 42 @Slf4j
40 43 public class JsonCoapAdaptor implements CoapTransportAdaptor {
41 44
42 45 @Override
43 46 public TransportProtos.PostTelemetryMsg convertToPostTelemetry(UUID sessionId, Request inbound) throws AdaptorException {
44   - String payload = validatePayload(sessionId, inbound);
  47 + String payload = validatePayload(sessionId, inbound, false);
45 48 try {
46 49 return JsonConverter.convertToTelemetryProto(new JsonParser().parse(payload));
47 50 } catch (IllegalStateException | JsonSyntaxException ex) {
... ... @@ -51,7 +54,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
51 54
52 55 @Override
53 56 public TransportProtos.PostAttributeMsg convertToPostAttributes(UUID sessionId, Request inbound) throws AdaptorException {
54   - String payload = validatePayload(sessionId, inbound);
  57 + String payload = validatePayload(sessionId, inbound, false);
55 58 try {
56 59 return JsonConverter.convertToAttributesProto(new JsonParser().parse(payload));
57 60 } catch (IllegalStateException | JsonSyntaxException ex) {
... ... @@ -79,7 +82,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
79 82 @Override
80 83 public TransportProtos.ToDeviceRpcResponseMsg convertToDeviceRpcResponse(UUID sessionId, Request inbound) throws AdaptorException {
81 84 Optional<Integer> requestId = CoapTransportResource.getRequestId(inbound);
82   - String payload = validatePayload(sessionId, inbound);
  85 + String payload = validatePayload(sessionId, inbound, false);
83 86 JsonObject response = new JsonParser().parse(payload).getAsJsonObject();
84 87 return TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId.orElseThrow(() -> new AdaptorException("Request id is missing!")))
85 88 .setPayload(response.toString()).build();
... ... @@ -87,11 +90,22 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
87 90
88 91 @Override
89 92 public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(UUID sessionId, Request inbound) throws AdaptorException {
90   - String payload = validatePayload(sessionId, inbound);
  93 + String payload = validatePayload(sessionId, inbound, false);
91 94 return JsonConverter.convertToServerRpcRequest(new JsonParser().parse(payload), 0);
92 95 }
93 96
94 97 @Override
  98 + public TransportProtos.ClaimDeviceMsg convertToClaimDevice(UUID sessionId, Request inbound, TransportProtos.SessionInfoProto sessionInfo) throws AdaptorException {
  99 + DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
  100 + String payload = validatePayload(sessionId, inbound, true);
  101 + try {
  102 + return JsonConverter.convertToClaimDeviceProto(deviceId, payload);
  103 + } catch (IllegalStateException | JsonSyntaxException ex) {
  104 + throw new AdaptorException(ex);
  105 + }
  106 + }
  107 +
  108 + @Override
95 109 public Response convertToPublish(CoapTransportResource.CoapSessionListener session, TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException {
96 110 return getObserveNotification(session.getNextSeqNumber(), JsonConverter.toJson(msg));
97 111 }
... ... @@ -128,11 +142,13 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
128 142 return response;
129 143 }
130 144
131   - private String validatePayload(UUID sessionId, Request inbound) throws AdaptorException {
  145 + private String validatePayload(UUID sessionId, Request inbound, boolean isEmptyPayloadAllowed) throws AdaptorException {
132 146 String payload = inbound.getPayloadString();
133 147 if (payload == null) {
134 148 log.warn("[{}] Payload is empty!", sessionId);
135   - throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
  149 + if (!isEmptyPayloadAllowed) {
  150 + throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
  151 + }
136 152 }
137 153 return payload;
138 154 }
... ...
... ... @@ -20,7 +20,6 @@ import com.google.gson.JsonParser;
20 20 import lombok.extern.slf4j.Slf4j;
21 21 import org.springframework.beans.factory.annotation.Autowired;
22 22 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
23   -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
24 23 import org.springframework.http.HttpStatus;
25 24 import org.springframework.http.ResponseEntity;
26 25 import org.springframework.util.StringUtils;
... ... @@ -31,6 +30,7 @@ import org.springframework.web.bind.annotation.RequestMethod;
31 30 import org.springframework.web.bind.annotation.RequestParam;
32 31 import org.springframework.web.bind.annotation.RestController;
33 32 import org.springframework.web.context.request.async.DeferredResult;
  33 +import org.thingsboard.server.common.data.id.DeviceId;
34 34 import org.thingsboard.server.common.transport.SessionMsgListener;
35 35 import org.thingsboard.server.common.transport.TransportContext;
36 36 import org.thingsboard.server.common.transport.TransportService;
... ... @@ -119,6 +119,20 @@ public class DeviceApiController {
119 119 return responseWriter;
120 120 }
121 121
  122 + @RequestMapping(value = "/{deviceToken}/claim", method = RequestMethod.POST)
  123 + public DeferredResult<ResponseEntity> claimDevice(@PathVariable("deviceToken") String deviceToken,
  124 + @RequestBody(required = false) String json, HttpServletRequest request) {
  125 + DeferredResult<ResponseEntity> responseWriter = new DeferredResult<>();
  126 + transportContext.getTransportService().process(ValidateDeviceTokenRequestMsg.newBuilder().setToken(deviceToken).build(),
  127 + new DeviceAuthCallback(transportContext, responseWriter, sessionInfo -> {
  128 + TransportService transportService = transportContext.getTransportService();
  129 + DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
  130 + transportService.process(sessionInfo, JsonConverter.convertToClaimDeviceProto(deviceId, json),
  131 + new HttpOkCallback(responseWriter));
  132 + }));
  133 + return responseWriter;
  134 + }
  135 +
122 136 @RequestMapping(value = "/{deviceToken}/rpc", method = RequestMethod.GET, produces = "application/json")
123 137 public DeferredResult<ResponseEntity> subscribeToCommands(@PathVariable("deviceToken") String deviceToken,
124 138 @RequestParam(value = "timeout", required = false, defaultValue = "0") long timeout,
... ...
... ... @@ -29,6 +29,7 @@ public class MqttTopics {
29 29 public static final String DEVICE_ATTRIBUTES_RESPONSES_TOPIC = DEVICE_ATTRIBUTES_RESPONSE_TOPIC_PREFIX + "+";
30 30 public static final String DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX = BASE_DEVICE_API_TOPIC + "/attributes/request/";
31 31 public static final String DEVICE_TELEMETRY_TOPIC = BASE_DEVICE_API_TOPIC + "/telemetry";
  32 + public static final String DEVICE_CLAIM_TOPIC = BASE_DEVICE_API_TOPIC + "/claim";
32 33 public static final String DEVICE_ATTRIBUTES_TOPIC = BASE_DEVICE_API_TOPIC + "/attributes";
33 34
34 35 public static final String BASE_GATEWAY_API_TOPIC = "v1/gateway";
... ... @@ -36,6 +37,7 @@ public class MqttTopics {
36 37 public static final String GATEWAY_DISCONNECT_TOPIC = BASE_GATEWAY_API_TOPIC + "/disconnect";
37 38 public static final String GATEWAY_ATTRIBUTES_TOPIC = BASE_GATEWAY_API_TOPIC + "/attributes";
38 39 public static final String GATEWAY_TELEMETRY_TOPIC = BASE_GATEWAY_API_TOPIC + "/telemetry";
  40 + public static final String GATEWAY_CLAIM_TOPIC = BASE_GATEWAY_API_TOPIC + "/claim";
39 41 public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + "/rpc";
40 42 public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + "/attributes/request";
41 43 public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + "/attributes/response";
... ...
... ... @@ -38,11 +38,11 @@ import io.netty.util.concurrent.Future;
38 38 import io.netty.util.concurrent.GenericFutureListener;
39 39 import lombok.extern.slf4j.Slf4j;
40 40 import org.springframework.util.StringUtils;
  41 +import org.thingsboard.server.common.msg.EncryptionUtil;
41 42 import org.thingsboard.server.common.transport.SessionMsgListener;
42 43 import org.thingsboard.server.common.transport.TransportService;
43 44 import org.thingsboard.server.common.transport.TransportServiceCallback;
44 45 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
45   -import org.thingsboard.server.common.msg.EncryptionUtil;
46 46 import org.thingsboard.server.common.transport.service.AbstractTransportService;
47 47 import org.thingsboard.server.gen.transport.TransportProtos;
48 48 import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto;
... ... @@ -183,6 +183,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
183 183 case MqttTopics.GATEWAY_TELEMETRY_TOPIC:
184 184 gatewaySessionHandler.onDeviceTelemetry(mqttMsg);
185 185 break;
  186 + case MqttTopics.GATEWAY_CLAIM_TOPIC:
  187 + gatewaySessionHandler.onDeviceClaim(mqttMsg);
  188 + break;
186 189 case MqttTopics.GATEWAY_ATTRIBUTES_TOPIC:
187 190 gatewaySessionHandler.onDeviceAttributes(mqttMsg);
188 191 break;
... ... @@ -221,6 +224,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
221 224 } else if (topicName.startsWith(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC)) {
222 225 TransportProtos.ToServerRpcRequestMsg rpcRequestMsg = adaptor.convertToServerRpcRequest(deviceSessionCtx, mqttMsg);
223 226 transportService.process(sessionInfo, rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
  227 + } else if (topicName.equals(MqttTopics.DEVICE_CLAIM_TOPIC)) {
  228 + TransportProtos.ClaimDeviceMsg claimDeviceMsg = adaptor.convertToClaimDevice(deviceSessionCtx, mqttMsg);
  229 + transportService.process(sessionInfo, claimDeviceMsg, getPubAckCallback(ctx, msgId, claimDeviceMsg));
224 230 }
225 231 } catch (AdaptorException e) {
226 232 log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
... ...
... ... @@ -57,7 +57,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
57 57
58 58 @Override
59 59 public TransportProtos.PostTelemetryMsg convertToPostTelemetry(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
60   - String payload = validatePayload(ctx.getSessionId(), inbound.payload());
  60 + String payload = validatePayload(ctx.getSessionId(), inbound.payload(), false);
61 61 try {
62 62 return JsonConverter.convertToTelemetryProto(new JsonParser().parse(payload));
63 63 } catch (IllegalStateException | JsonSyntaxException ex) {
... ... @@ -67,7 +67,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
67 67
68 68 @Override
69 69 public TransportProtos.PostAttributeMsg convertToPostAttributes(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
70   - String payload = validatePayload(ctx.getSessionId(), inbound.payload());
  70 + String payload = validatePayload(ctx.getSessionId(), inbound.payload(), false);
71 71 try {
72 72 return JsonConverter.convertToAttributesProto(new JsonParser().parse(payload));
73 73 } catch (IllegalStateException | JsonSyntaxException ex) {
... ... @@ -114,7 +114,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
114 114 @Override
115 115 public TransportProtos.ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
116 116 String topicName = inbound.variableHeader().topicName();
117   - String payload = validatePayload(ctx.getSessionId(), inbound.payload());
  117 + String payload = validatePayload(ctx.getSessionId(), inbound.payload(), false);
118 118 try {
119 119 Integer requestId = Integer.valueOf(topicName.substring(MqttTopics.DEVICE_RPC_REQUESTS_TOPIC.length()));
120 120 return JsonConverter.convertToServerRpcRequest(new JsonParser().parse(payload), requestId);
... ... @@ -124,6 +124,16 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
124 124 }
125 125
126 126 @Override
  127 + public TransportProtos.ClaimDeviceMsg convertToClaimDevice(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
  128 + String payload = validatePayload(ctx.getSessionId(), inbound.payload(), true);
  129 + try {
  130 + return JsonConverter.convertToClaimDeviceProto(ctx.getDeviceId(), payload);
  131 + } catch (IllegalStateException | JsonSyntaxException ex) {
  132 + throw new AdaptorException(ex);
  133 + }
  134 + }
  135 +
  136 + @Override
127 137 public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
128 138 if (!StringUtils.isEmpty(responseMsg.getError())) {
129 139 throw new AdaptorException(responseMsg.getError());
... ... @@ -193,7 +203,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
193 203 }
194 204
195 205 public static JsonElement validateJsonPayload(UUID sessionId, ByteBuf payloadData) throws AdaptorException {
196   - String payload = validatePayload(sessionId, payloadData);
  206 + String payload = validatePayload(sessionId, payloadData, false);
197 207 try {
198 208 return new JsonParser().parse(payload);
199 209 } catch (JsonSyntaxException ex) {
... ... @@ -201,12 +211,14 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
201 211 }
202 212 }
203 213
204   - private static String validatePayload(UUID sessionId, ByteBuf payloadData) throws AdaptorException {
  214 + private static String validatePayload(UUID sessionId, ByteBuf payloadData, boolean isEmptyPayloadAllowed) throws AdaptorException {
205 215 try {
206 216 String payload = payloadData.toString(UTF8);
207 217 if (payload == null) {
208 218 log.warn("[{}] Payload is empty!", sessionId);
209   - throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
  219 + if (!isEmptyPayloadAllowed) {
  220 + throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
  221 + }
210 222 }
211 223 return payload;
212 224 } finally {
... ...
... ... @@ -19,6 +19,7 @@ import io.netty.handler.codec.mqtt.MqttMessage;
19 19 import io.netty.handler.codec.mqtt.MqttPublishMessage;
20 20 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
21 21 import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
  22 +import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
22 23 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
23 24 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg;
24 25 import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
... ... @@ -46,6 +47,8 @@ public interface MqttTransportAdaptor {
46 47
47 48 ToServerRpcRequestMsg convertToServerRpcRequest(MqttDeviceAwareSessionContext ctx, MqttPublishMessage mqttMsg) throws AdaptorException;
48 49
  50 + ClaimDeviceMsg convertToClaimDevice(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException;
  51 +
49 52 Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, GetAttributeResponseMsg responseMsg) throws AdaptorException;
50 53
51 54 Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, GetAttributeResponseMsg responseMsg) throws AdaptorException;
... ... @@ -59,4 +62,5 @@ public interface MqttTransportAdaptor {
59 62 Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException;
60 63
61 64 Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, ToServerRpcResponseMsg rpcResponse) throws AdaptorException;
  65 +
62 66 }
... ...
... ... @@ -30,6 +30,7 @@ import io.netty.handler.codec.mqtt.MqttMessage;
30 30 import io.netty.handler.codec.mqtt.MqttPublishMessage;
31 31 import lombok.extern.slf4j.Slf4j;
32 32 import org.springframework.util.StringUtils;
  33 +import org.thingsboard.server.common.data.id.DeviceId;
33 34 import org.thingsboard.server.common.transport.TransportService;
34 35 import org.thingsboard.server.common.transport.TransportServiceCallback;
35 36 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
... ... @@ -183,7 +184,42 @@ public class GatewaySessionHandler {
183 184
184 185 @Override
185 186 public void onFailure(Throwable t) {
186   - log.debug("[{}] Failed to process device teleemtry command: {}", sessionId, deviceName, t);
  187 + log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t);
  188 + }
  189 + }, context.getExecutor());
  190 + }
  191 + } else {
  192 + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
  193 + }
  194 + }
  195 +
  196 + public void onDeviceClaim(MqttPublishMessage mqttMsg) throws AdaptorException {
  197 + JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, mqttMsg.payload());
  198 + int msgId = mqttMsg.variableHeader().packetId();
  199 + if (json.isJsonObject()) {
  200 + JsonObject jsonObj = json.getAsJsonObject();
  201 + for (Map.Entry<String, JsonElement> deviceEntry : jsonObj.entrySet()) {
  202 + String deviceName = deviceEntry.getKey();
  203 + Futures.addCallback(checkDeviceConnected(deviceName),
  204 + new FutureCallback<GatewayDeviceSessionCtx>() {
  205 + @Override
  206 + public void onSuccess(@Nullable GatewayDeviceSessionCtx deviceCtx) {
  207 + if (!deviceEntry.getValue().isJsonObject()) {
  208 + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
  209 + }
  210 + try {
  211 + DeviceId deviceId = deviceCtx.getDeviceId();
  212 + TransportProtos.ClaimDeviceMsg claimDeviceMsg = JsonConverter.convertToClaimDeviceProto(deviceId, deviceEntry.getValue());
  213 + transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg));
  214 + } catch (Throwable e) {
  215 + UUID gatewayId = new UUID(gateway.getDeviceIdMSB(), gateway.getDeviceIdLSB());
  216 + log.warn("[{}][{}] Failed to convert claim message: {}", gatewayId, deviceName, deviceEntry.getValue(), e);
  217 + }
  218 + }
  219 +
  220 + @Override
  221 + public void onFailure(Throwable t) {
  222 + log.debug("[{}] Failed to process device claiming command: {}", sessionId, deviceName, t);
187 223 }
188 224 }, context.getExecutor());
189 225 }
... ... @@ -209,6 +245,7 @@ public class GatewaySessionHandler {
209 245 TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(deviceEntry.getValue().getAsJsonObject());
210 246 transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg));
211 247 }
  248 +
212 249 @Override
213 250 public void onFailure(Throwable t) {
214 251 log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t);
... ...
... ... @@ -16,18 +16,19 @@
16 16 package org.thingsboard.server.common.transport;
17 17
18 18 import org.thingsboard.server.gen.transport.TransportProtos;
19   -import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
20   -import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
21   -import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
22   -import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
23   -import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
  19 +import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
  20 +import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
24 21 import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
25 22 import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
26 23 import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg;
  24 +import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
  25 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
  26 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
  27 +import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
  28 +import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
27 29 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg;
28 30 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
29 31 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
30   -import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
31 32
32 33 /**
33 34 * Created by ashvayka on 04.10.18.
... ... @@ -63,6 +64,8 @@ public interface TransportService {
63 64
64 65 void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SubscriptionInfoProto msg, TransportServiceCallback<Void> callback);
65 66
  67 + void process(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback<Void> callback);
  68 +
66 69 void registerAsyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener);
67 70
68 71 void registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout);
... ...
... ... @@ -24,6 +24,8 @@ import com.google.gson.JsonPrimitive;
24 24 import com.google.gson.JsonSyntaxException;
25 25 import org.apache.commons.lang3.math.NumberUtils;
26 26 import org.springframework.util.StringUtils;
  27 +import org.thingsboard.server.common.data.DataConstants;
  28 +import org.thingsboard.server.common.data.id.DeviceId;
27 29 import org.thingsboard.server.common.data.kv.AttributeKey;
28 30 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
29 31 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
... ... @@ -35,6 +37,7 @@ import org.thingsboard.server.common.data.kv.StringDataEntry;
35 37 import org.thingsboard.server.common.msg.kv.AttributesKVMsg;
36 38 import org.thingsboard.server.gen.transport.TransportProtos;
37 39 import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
  40 +import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
38 41 import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg;
39 42 import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
40 43 import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
... ... @@ -63,23 +66,69 @@ public class JsonConverter {
63 66
64 67 private static int maxStringValueLength = 0;
65 68
66   - public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonObject) throws JsonSyntaxException {
67   - long systemTs = System.currentTimeMillis();
  69 + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement) throws JsonSyntaxException {
68 70 PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder();
69   - if (jsonObject.isJsonObject()) {
70   - parseObject(builder, systemTs, jsonObject);
71   - } else if (jsonObject.isJsonArray()) {
72   - jsonObject.getAsJsonArray().forEach(je -> {
  71 + convertToTelemetry(jsonElement, System.currentTimeMillis(), null, builder);
  72 + return builder.build();
  73 + }
  74 +
  75 + private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map<Long, List<KvEntry>> result, PostTelemetryMsg.Builder builder) {
  76 + if (jsonElement.isJsonObject()) {
  77 + parseObject(systemTs, result, builder, jsonElement.getAsJsonObject());
  78 + } else if (jsonElement.isJsonArray()) {
  79 + jsonElement.getAsJsonArray().forEach(je -> {
73 80 if (je.isJsonObject()) {
74   - parseObject(builder, systemTs, je.getAsJsonObject());
  81 + parseObject(systemTs, result, builder, je.getAsJsonObject());
75 82 } else {
76 83 throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je);
77 84 }
78 85 });
79 86 } else {
80   - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonObject);
  87 + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonElement);
81 88 }
82   - return builder.build();
  89 + }
  90 +
  91 + private static void parseObject(long systemTs, Map<Long, List<KvEntry>> result, PostTelemetryMsg.Builder builder, JsonObject jo) {
  92 + if (result != null) {
  93 + parseObject(result, systemTs, jo);
  94 + } else {
  95 + parseObject(builder, systemTs, jo);
  96 + }
  97 + }
  98 +
  99 + public static ClaimDeviceMsg convertToClaimDeviceProto(DeviceId deviceId, String json) {
  100 + long durationMs = 0L;
  101 + if (json != null && !json.isEmpty()) {
  102 + return convertToClaimDeviceProto(deviceId, new JsonParser().parse(json));
  103 + }
  104 + return buildClaimDeviceMsg(deviceId, DataConstants.DEFAULT_SECRET_KEY, durationMs);
  105 + }
  106 +
  107 + public static ClaimDeviceMsg convertToClaimDeviceProto(DeviceId deviceId, JsonElement jsonElement) {
  108 + String secretKey = DataConstants.DEFAULT_SECRET_KEY;
  109 + long durationMs = 0L;
  110 + if (jsonElement.isJsonObject()) {
  111 + JsonObject jo = jsonElement.getAsJsonObject();
  112 + if (jo.has(DataConstants.SECRET_KEY_FIELD_NAME)) {
  113 + secretKey = jo.get(DataConstants.SECRET_KEY_FIELD_NAME).getAsString();
  114 + }
  115 + if (jo.has(DataConstants.DURATION_MS_FIELD_NAME)) {
  116 + durationMs = jo.get(DataConstants.DURATION_MS_FIELD_NAME).getAsLong();
  117 + }
  118 + } else {
  119 + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonElement);
  120 + }
  121 + return buildClaimDeviceMsg(deviceId, secretKey, durationMs);
  122 + }
  123 +
  124 + private static ClaimDeviceMsg buildClaimDeviceMsg(DeviceId deviceId, String secretKey, long durationMs) {
  125 + ClaimDeviceMsg.Builder result = ClaimDeviceMsg.newBuilder();
  126 + return result
  127 + .setDeviceIdMSB(deviceId.getId().getMostSignificantBits())
  128 + .setDeviceIdLSB(deviceId.getId().getLeastSignificantBits())
  129 + .setSecretKey(secretKey)
  130 + .setDurationMs(durationMs)
  131 + .build();
83 132 }
84 133
85 134 public static PostAttributeMsg convertToAttributesProto(JsonElement jsonObject) throws JsonSyntaxException {
... ... @@ -103,8 +152,7 @@ public class JsonConverter {
103 152 return result;
104 153 }
105 154
106   - private static void parseObject(PostTelemetryMsg.Builder builder, long systemTs, JsonElement jsonObject) {
107   - JsonObject jo = jsonObject.getAsJsonObject();
  155 + private static void parseObject(PostTelemetryMsg.Builder builder, long systemTs, JsonObject jo) {
108 156 if (jo.has("ts") && jo.has("values")) {
109 157 parseWithTs(builder, jo);
110 158 } else {
... ... @@ -137,7 +185,7 @@ public class JsonConverter {
137 185 String message = String.format("String value length [%d] for key [%s] is greater than maximum allowed [%d]", value.getAsString().length(), valueEntry.getKey(), maxStringValueLength);
138 186 throw new JsonSyntaxException(message);
139 187 }
140   - if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
  188 + if (isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
141 189 try {
142 190 result.add(buildNumericKeyValueProto(value, valueEntry.getKey()));
143 191 } catch (RuntimeException th) {
... ... @@ -400,7 +448,7 @@ public class JsonConverter {
400 448 String message = String.format("String value length [%d] for key [%s] is greater than maximum allowed [%d]", value.getAsString().length(), valueEntry.getKey(), maxStringValueLength);
401 449 throw new JsonSyntaxException(message);
402 450 }
403   - if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
  451 + if (isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
404 452 try {
405 453 parseNumericValue(result, valueEntry, value);
406 454 } catch (RuntimeException th) {
... ... @@ -423,26 +471,13 @@ public class JsonConverter {
423 471 return result;
424 472 }
425 473
426   - public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonObject, long systemTs) throws JsonSyntaxException {
  474 + public static Map<Long, List<KvEntry>> convertToTelemetry(JsonElement jsonElement, long systemTs) throws JsonSyntaxException {
427 475 Map<Long, List<KvEntry>> result = new HashMap<>();
428   - if (jsonObject.isJsonObject()) {
429   - parseObject(result, systemTs, jsonObject);
430   - } else if (jsonObject.isJsonArray()) {
431   - jsonObject.getAsJsonArray().forEach(je -> {
432   - if (je.isJsonObject()) {
433   - parseObject(result, systemTs, je.getAsJsonObject());
434   - } else {
435   - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je);
436   - }
437   - });
438   - } else {
439   - throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonObject);
440   - }
  476 + convertToTelemetry(jsonElement, systemTs, result, null);
441 477 return result;
442 478 }
443 479
444   - private static void parseObject(Map<Long, List<KvEntry>> result, long systemTs, JsonElement jsonObject) {
445   - JsonObject jo = jsonObject.getAsJsonObject();
  480 + private static void parseObject(Map<Long, List<KvEntry>> result, long systemTs, JsonObject jo) {
446 481 if (jo.has("ts") && jo.has("values")) {
447 482 parseWithTs(result, jo);
448 483 } else {
... ...
... ... @@ -28,7 +28,12 @@ import org.thingsboard.server.common.transport.TransportServiceCallback;
28 28 import org.thingsboard.server.gen.transport.TransportProtos;
29 29
30 30 import java.util.UUID;
31   -import java.util.concurrent.*;
  31 +import java.util.concurrent.ConcurrentHashMap;
  32 +import java.util.concurrent.ConcurrentMap;
  33 +import java.util.concurrent.ExecutorService;
  34 +import java.util.concurrent.Executors;
  35 +import java.util.concurrent.ScheduledExecutorService;
  36 +import java.util.concurrent.TimeUnit;
32 37
33 38 /**
34 39 * Created by ashvayka on 17.10.18.
... ... @@ -128,6 +133,12 @@ public abstract class AbstractTransportService implements TransportService {
128 133 }
129 134
130 135 @Override
  136 + public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ClaimDeviceMsg msg,
  137 + TransportServiceCallback<Void> callback) {
  138 + registerClaimingInfo(sessionInfo, msg, callback);
  139 + }
  140 +
  141 + @Override
131 142 public void reportActivity(TransportProtos.SessionInfoProto sessionInfo) {
132 143 reportActivityInternal(sessionInfo);
133 144 }
... ... @@ -148,6 +159,8 @@ public abstract class AbstractTransportService implements TransportService {
148 159
149 160 protected abstract void doProcess(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback);
150 161
  162 + protected abstract void registerClaimingInfo(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ClaimDeviceMsg msg, TransportServiceCallback<Void> callback);
  163 +
151 164 private SessionMetaData reportActivityInternal(TransportProtos.SessionInfoProto sessionInfo) {
152 165 UUID sessionId = toId(sessionInfo);
153 166 SessionMetaData sessionMetaData = sessions.get(sessionId);
... ...
... ... @@ -24,34 +24,40 @@ import org.apache.kafka.clients.producer.RecordMetadata;
24 24 import org.springframework.beans.factory.annotation.Autowired;
25 25 import org.springframework.beans.factory.annotation.Value;
26 26 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
27   -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
28 27 import org.springframework.stereotype.Service;
29   -import org.thingsboard.server.common.transport.SessionMsgListener;
30   -import org.thingsboard.server.common.transport.TransportService;
31 28 import org.thingsboard.server.common.transport.TransportServiceCallback;
32   -import org.thingsboard.server.gen.transport.TransportProtos;
33   -import org.thingsboard.server.gen.transport.TransportProtos.*;
  29 +import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
  30 +import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
34 31 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg;
35 32 import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg;
36 33 import org.thingsboard.server.gen.transport.TransportProtos.PostAttributeMsg;
37 34 import org.thingsboard.server.gen.transport.TransportProtos.PostTelemetryMsg;
38 35 import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg;
39 36 import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
  37 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
  38 +import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
  39 +import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionInfoProto;
  40 +import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
  41 +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  42 +import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcRequestMsg;
40 43 import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
41 44 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg;
42 45 import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg;
43   -import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
  46 +import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg;
44 47 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCredentialsResponseMsg;
45 48 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenRequestMsg;
46 49 import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
47   -import org.thingsboard.server.kafka.*;
  50 +import org.thingsboard.server.kafka.AsyncCallbackTemplate;
  51 +import org.thingsboard.server.kafka.TBKafkaAdmin;
  52 +import org.thingsboard.server.kafka.TBKafkaConsumerTemplate;
  53 +import org.thingsboard.server.kafka.TBKafkaProducerTemplate;
  54 +import org.thingsboard.server.kafka.TbKafkaRequestTemplate;
  55 +import org.thingsboard.server.kafka.TbKafkaSettings;
  56 +import org.thingsboard.server.kafka.TbNodeIdProvider;
48 57
49 58 import javax.annotation.PostConstruct;
50 59 import javax.annotation.PreDestroy;
51 60 import java.time.Duration;
52   -import java.util.UUID;
53   -import java.util.concurrent.ConcurrentHashMap;
54   -import java.util.concurrent.ConcurrentMap;
55 61 import java.util.concurrent.ExecutorService;
56 62 import java.util.concurrent.Executors;
57 63
... ... @@ -305,6 +311,15 @@ public class RemoteTransportService extends AbstractTransportService {
305 311 send(sessionInfo, toRuleEngineMsg, callback);
306 312 }
307 313
  314 + @Override
  315 + protected void registerClaimingInfo(SessionInfoProto sessionInfo, ClaimDeviceMsg msg, TransportServiceCallback<Void> callback) {
  316 + ToRuleEngineMsg toRuleEngineMsg = ToRuleEngineMsg.newBuilder().setToDeviceActorMsg(
  317 + TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
  318 + .setClaimDevice(msg).build()
  319 + ).build();
  320 + send(sessionInfo, toRuleEngineMsg, callback);
  321 + }
  322 +
308 323 private static class TransportCallbackAdaptor implements Callback {
309 324 private final TransportServiceCallback<Void> callback;
310 325
... ...
... ... @@ -172,6 +172,13 @@ message ToServerRpcResponseMsg {
172 172 string error = 3;
173 173 }
174 174
  175 +message ClaimDeviceMsg {
  176 + int64 deviceIdMSB = 1;
  177 + int64 deviceIdLSB = 2;
  178 + string secretKey = 3;
  179 + int64 durationMs = 4;
  180 +}
  181 +
175 182 //Used to report session state to tb-node and persist this state in the cache on the tb-node level.
176 183 message SubscriptionInfoProto {
177 184 int64 lastActivityTime = 1;
... ... @@ -199,6 +206,7 @@ message TransportToDeviceActorMsg {
199 206 ToDeviceRpcResponseMsg toDeviceRPCCallResponse = 8;
200 207 ToServerRpcRequestMsg toServerRPCCallRequest = 9;
201 208 SubscriptionInfoProto subscriptionInfo = 10;
  209 + ClaimDeviceMsg claimDevice = 11;
202 210 }
203 211
204 212 message DeviceActorToTransportMsg {
... ...
  1 +/**
  2 + * Copyright © 2016-2019 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.device;
  17 +
  18 +import com.google.common.util.concurrent.ListenableFuture;
  19 +import org.thingsboard.server.common.data.Device;
  20 +import org.thingsboard.server.common.data.id.CustomerId;
  21 +import org.thingsboard.server.common.data.id.DeviceId;
  22 +import org.thingsboard.server.common.data.id.TenantId;
  23 +import org.thingsboard.server.dao.device.claim.ClaimResponse;
  24 +
  25 +import java.util.List;
  26 +
  27 +public interface ClaimDevicesService {
  28 +
  29 + ListenableFuture<Void> registerClaimingInfo(TenantId tenantId, DeviceId deviceId, String secretKey, long durationMs);
  30 +
  31 + ListenableFuture<ClaimResponse> claimDevice(Device device, CustomerId customerId, String secretKey);
  32 +
  33 + ListenableFuture<List<Void>> reClaimDevice(TenantId tenantId, Device device);
  34 +
  35 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2019 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.device;
  17 +
  18 +import com.google.common.util.concurrent.Futures;
  19 +import com.google.common.util.concurrent.ListenableFuture;
  20 +import lombok.extern.slf4j.Slf4j;
  21 +import org.springframework.beans.factory.annotation.Autowired;
  22 +import org.springframework.beans.factory.annotation.Value;
  23 +import org.springframework.cache.Cache;
  24 +import org.springframework.cache.CacheManager;
  25 +import org.springframework.stereotype.Service;
  26 +import org.thingsboard.server.common.data.DataConstants;
  27 +import org.thingsboard.server.common.data.Device;
  28 +import org.thingsboard.server.common.data.id.CustomerId;
  29 +import org.thingsboard.server.common.data.id.DeviceId;
  30 +import org.thingsboard.server.common.data.id.TenantId;
  31 +import org.thingsboard.server.common.data.kv.AttributeKvEntry;
  32 +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
  33 +import org.thingsboard.server.common.data.kv.BooleanDataEntry;
  34 +import org.thingsboard.server.dao.attributes.AttributesService;
  35 +import org.thingsboard.server.dao.device.claim.ClaimData;
  36 +import org.thingsboard.server.dao.device.claim.ClaimResponse;
  37 +import org.thingsboard.server.dao.model.ModelConstants;
  38 +
  39 +import java.util.Collections;
  40 +import java.util.List;
  41 +import java.util.Optional;
  42 +
  43 +import static org.thingsboard.server.common.data.CacheConstants.CLAIM_DEVICES_CACHE;
  44 +
  45 +@Service
  46 +@Slf4j
  47 +public class ClaimDevicesServiceImpl implements ClaimDevicesService {
  48 +
  49 + private static final String CLAIM_ATTRIBUTE_NAME = "claimingAllowed";
  50 +
  51 + @Autowired
  52 + private DeviceService deviceService;
  53 + @Autowired
  54 + private AttributesService attributesService;
  55 + @Autowired
  56 + private CacheManager cacheManager;
  57 +
  58 + @Value("${security.claim.allowClaimingByDefault}")
  59 + private boolean isAllowedClaimingByDefault;
  60 +
  61 + @Value("${security.claim.duration}")
  62 + private long systemDurationMs;
  63 +
  64 + @Override
  65 + public ListenableFuture<Void> registerClaimingInfo(TenantId tenantId, DeviceId deviceId, String secretKey, long durationMs) {
  66 + ListenableFuture<Device> deviceFuture = deviceService.findDeviceByIdAsync(tenantId, deviceId);
  67 + return Futures.transformAsync(deviceFuture, device -> {
  68 + Cache cache = cacheManager.getCache(CLAIM_DEVICES_CACHE);
  69 + List<Object> key = constructCacheKey(device.getId());
  70 +
  71 + if (isAllowedClaimingByDefault) {
  72 + if (device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
  73 + persistInCache(secretKey, durationMs, cache, key);
  74 + return Futures.immediateFuture(null);
  75 + }
  76 + log.warn("The device [{}] has been already claimed!", device.getName());
  77 + throw new IllegalArgumentException();
  78 + } else {
  79 + ListenableFuture<List<AttributeKvEntry>> claimingAllowedFuture = attributesService.find(tenantId, device.getId(),
  80 + DataConstants.SERVER_SCOPE, Collections.singletonList(CLAIM_ATTRIBUTE_NAME));
  81 + return Futures.transform(claimingAllowedFuture, list -> {
  82 + if (list != null && !list.isEmpty()) {
  83 + Optional<Boolean> claimingAllowedOptional = list.get(0).getBooleanValue();
  84 + if (claimingAllowedOptional.isPresent() && claimingAllowedOptional.get()
  85 + && device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
  86 + persistInCache(secretKey, durationMs, cache, key);
  87 + return null;
  88 + }
  89 + }
  90 + log.warn("Failed to find claimingAllowed attribute for device or it is already claimed![{}]", device.getName());
  91 + throw new IllegalArgumentException();
  92 + });
  93 + }
  94 + });
  95 + }
  96 +
  97 + @Override
  98 + public ListenableFuture<ClaimResponse> claimDevice(Device device, CustomerId customerId, String secretKey) {
  99 + List<Object> key = constructCacheKey(device.getId());
  100 + Cache cache = cacheManager.getCache(CLAIM_DEVICES_CACHE);
  101 + ClaimData claimData = cache.get(key, ClaimData.class);
  102 + if (claimData != null) {
  103 + long currTs = System.currentTimeMillis();
  104 + if (currTs > claimData.getExpirationTime() || !secretKey.equals(claimData.getSecretKey())) {
  105 + log.warn("The claiming timeout occurred or wrong 'secretKey' provided for the device [{}]", device.getName());
  106 + cache.evict(key);
  107 + return Futures.immediateFuture(ClaimResponse.FAILURE);
  108 + } else {
  109 + if (device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
  110 + device.setCustomerId(customerId);
  111 + deviceService.saveDevice(device);
  112 + return Futures.transform(removeClaimingSavedData(cache, key, device), result -> ClaimResponse.SUCCESS);
  113 + }
  114 + return Futures.transform(removeClaimingSavedData(cache, key, device), result -> ClaimResponse.CLAIMED);
  115 + }
  116 + } else {
  117 + log.warn("Failed to find the device's claiming message![{}]", device.getName());
  118 + return Futures.immediateFuture(ClaimResponse.CLAIMED);
  119 + }
  120 + }
  121 +
  122 + @Override
  123 + public ListenableFuture<List<Void>> reClaimDevice(TenantId tenantId, Device device) {
  124 + if (!device.getCustomerId().getId().equals(ModelConstants.NULL_UUID)) {
  125 + cacheEviction(device.getId());
  126 +
  127 + device.setCustomerId(null);
  128 + deviceService.saveDevice(device);
  129 + if (isAllowedClaimingByDefault) {
  130 + return Futures.immediateFuture(Collections.emptyList());
  131 + }
  132 + return attributesService.save(tenantId, device.getId(), DataConstants.SERVER_SCOPE, Collections.singletonList(
  133 + new BaseAttributeKvEntry(new BooleanDataEntry(CLAIM_ATTRIBUTE_NAME, true),
  134 + System.currentTimeMillis())));
  135 + }
  136 + cacheEviction(device.getId());
  137 + return Futures.immediateFuture(Collections.emptyList());
  138 + }
  139 +
  140 + private List<Object> constructCacheKey(DeviceId deviceId) {
  141 + return Collections.singletonList(deviceId);
  142 + }
  143 +
  144 + private void persistInCache(String secretKey, long durationMs, Cache cache, List<Object> key) {
  145 + ClaimData claimData = new ClaimData(secretKey,
  146 + System.currentTimeMillis() + validateDurationMs(durationMs));
  147 + cache.putIfAbsent(key, claimData);
  148 + }
  149 +
  150 + private long validateDurationMs(long durationMs) {
  151 + if (durationMs > 0L) {
  152 + return durationMs;
  153 + }
  154 + return systemDurationMs;
  155 + }
  156 +
  157 + private ListenableFuture<List<Void>> removeClaimingSavedData(Cache cache, List<Object> key, Device device) {
  158 + cache.evict(key);
  159 + if (isAllowedClaimingByDefault) {
  160 + return Futures.immediateFuture(null);
  161 + }
  162 + return attributesService.removeAll(device.getTenantId(),
  163 + device.getId(), DataConstants.SERVER_SCOPE, Collections.singletonList(CLAIM_ATTRIBUTE_NAME));
  164 + }
  165 +
  166 + private void cacheEviction(DeviceId deviceId) {
  167 + Cache cache = cacheManager.getCache(CLAIM_DEVICES_CACHE);
  168 + cache.evict(constructCacheKey(deviceId));
  169 + }
  170 +
  171 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2019 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.device.claim;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.Data;
  20 +
  21 +@AllArgsConstructor
  22 +@Data
  23 +public class ClaimData {
  24 +
  25 + private final String secretKey;
  26 + private final long expirationTime;
  27 +
  28 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2019 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.device.claim;
  17 +
  18 +public enum ClaimResponse {
  19 +
  20 + SUCCESS,
  21 + FAILURE,
  22 + CLAIMED
  23 +
  24 +}
... ...
... ... @@ -27,11 +27,16 @@ caffeine.specs.assets.maxSize=100000
27 27 caffeine.specs.entityViews.timeToLiveInMinutes=1440
28 28 caffeine.specs.entityViews.maxSize=100000
29 29
  30 +caffeine.specs.claimDevices.timeToLiveInMinutes=1440
  31 +caffeine.specs.claimDevices.maxSize=100000
  32 +
30 33 redis.connection.host=localhost
31 34 redis.connection.port=6379
32 35 redis.connection.db=0
33 36 redis.connection.password=
34 37
35 38 security.user_login_case_sensitive=true
  39 +security.claim.allowClaimingByDefault=true
  40 +security.claim.duration=60000
36 41
37   -database.ts_max_intervals=700
\ No newline at end of file
  42 +database.ts_max_intervals=700
... ...