Skip to content

[To dev/1.3] Fix subscription topic authorization bypass (#18418) #18435

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: dev/1.3
Choose a base branch
from
Open

[To dev/1.3] Fix subscription topic authorization bypass (#18418) #18435

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2828,7 +2828,7 @@ public TPipeSubscribeResp pipeSubscribe(final TPipeSubscribeReq req) {
return getNotLoggedInPipeSubscribeResp();
}

return SubscriptionAgent.receiver().handle(req);
return SubscriptionAgent.receiver().handle(req, clientSession.getUsername());
} finally {
SESSION_MANAGER.updateIdleTime();
}
Expand Down
11 changes: 11 additions & 0 deletions ...anode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionReceiverAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,13 +77,24 @@ public class SubscriptionReceiverAgent {
}

public TPipeSubscribeResp handle(final TPipeSubscribeReq req) {
return handle(req, null);
}

public TPipeSubscribeResp handle(final TPipeSubscribeReq req, final String username) {
if (username == null) {
return new TPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.NO_PERMISSION),
PipeSubscribeResponseVersion.VERSION_1.getVersion(),
PipeSubscribeResponseType.ACK.getType());
}
if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
return SUBSCRIPTION_NOT_ENABLED_ERROR_RESP;
}

final byte reqVersion = req.getVersion();
if (RECEIVER_CONSTRUCTORS.containsKey(reqVersion)) {
final SubscriptionReceiver receiver = getReceiver(reqVersion);
receiver.setAuthenticatedUsername(username);
activeReceivers.add(receiver);
receiver.handleTimeout();
return receiver.handle(req);
Expand Down
55 changes: 55 additions & 0 deletions ...datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,18 @@

package org.apache.iotdb.db.subscription.agent;

import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.entity.PrivilegeType;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaRespExceptionMessage;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;

Expand All @@ -30,6 +39,7 @@

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -187,4 +197,49 @@ public Map<String, TopicConfig> getTopicConfigs(final Set<String> topicNames) {
releaseReadLock();
}
}

/**
* Check that the authenticated session can read all data covered by the requested topics. The
* username in ConsumerConfig is client-controlled and therefore must not be used as the
* authorization identity.
*/
public TSStatus checkTopicReadPermissions(
final String username, final Iterable<String> topicNames) {
if (Objects.isNull(username)) {
return RpcUtils.getStatus(TSStatusCode.NO_PERMISSION);
}

acquireReadLock();
try {
for (final String topicName : topicNames) {
final TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName);
if (Objects.isNull(topicMeta)) {
continue;
}

final TSStatus status = checkTopicReadPermission(username, topicMeta);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
}
}
return RpcUtils.SUCCESS_STATUS;
} finally {
releaseReadLock();
}
}

private TSStatus checkTopicReadPermission(final String username, final TopicMeta topicMeta) {
final TopicConfig topicConfig = topicMeta.getConfig();
final PipePattern pipePattern =
topicConfig.getAttribute().containsKey(TopicConstant.PATTERN_KEY)
? new PrefixPipePattern(topicConfig.getAttribute().get(TopicConstant.PATTERN_KEY))
: new IoTDBPipePattern(
topicConfig.getStringOrDefault(
TopicConstant.PATH_KEY, TopicConstant.PATH_DEFAULT_VALUE));
final List<PartialPath> paths = pipePattern.getBaseInclusionPaths();
return AuthorityChecker.getTSStatus(
AuthorityChecker.checkPatternPermission(username, paths, PrivilegeType.READ_DATA.ordinal()),
paths,
PrivilegeType.READ_DATA);
}
}
2 changes: 2 additions & 0 deletions ...atanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@ public interface SubscriptionReceiver {

TPipeSubscribeResp handle(TPipeSubscribeReq req);

void setAuthenticatedUsername(final String username);

PipeSubscribeRequestVersion getVersion();

void handleExit();
Expand Down
64 changes: 58 additions & 6 deletions ...anode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ public class SubscriptionReceiverV1 implements SubscriptionReceiver {

private final ThreadLocal<ConsumerConfig> consumerConfigThreadLocal = new ThreadLocal<>();
private final ThreadLocal<PollTimer> pollTimerThreadLocal = new ThreadLocal<>();
private volatile String authenticatedUsername;
private volatile ConsumerConfig sharedConsumerConfig;
private volatile boolean consumerInvalidated;
private volatile long lastActivityTimeMs = System.currentTimeMillis();
Expand Down Expand Up @@ -167,6 +168,11 @@ public PipeSubscribeRequestVersion getVersion() {
return PipeSubscribeRequestVersion.VERSION_1;
}

@Override
public void setAuthenticatedUsername(final String username) {
authenticatedUsername = username;
}

@Override
public void handleExit() {
final ConsumerConfig consumerConfig = consumerConfigThreadLocal.get();
Expand All @@ -183,6 +189,7 @@ public void handleExit() {
consumerConfigThreadLocal.remove();
}
clearSharedConsumerState();
authenticatedUsername = null;
}

@Override
Expand Down Expand Up @@ -322,17 +329,22 @@ private TPipeSubscribeResp handlePipeSubscribeHeartbeatInternal(
return SUBSCRIPTION_MISSING_CUSTOMER_RESP;
}

// TODO: do something
final Set<String> subscribedTopicNames =
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId());
final TSStatus readPermissionStatus =
SubscriptionAgent.topic()
.checkTopicReadPermissions(authenticatedUsername, subscribedTopicNames);
if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(readPermissionStatus);
}

LOGGER.info("Subscription: consumer {} heartbeat successfully", consumerConfig);

// fetch subscribed topics
final Map<String, TopicConfig> topics =
SubscriptionAgent.topic()
.getTopicConfigs(
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId()));
SubscriptionAgent.topic().getTopicConfigs(subscribedTopicNames);

// fetch available endpoints
final Map<Integer, TEndPoint> endPoints = new HashMap<>();
Expand Down Expand Up @@ -403,6 +415,11 @@ private TPipeSubscribeResp handlePipeSubscribeSubscribeInternal(

// subscribe topics
final Set<String> topicNames = req.getTopicNames();
final TSStatus readPermissionStatus =
SubscriptionAgent.topic().checkTopicReadPermissions(authenticatedUsername, topicNames);
if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(readPermissionStatus);
}
subscribe(consumerConfig, topicNames);

LOGGER.info("Subscription: consumer {} subscribe {} successfully", consumerConfig, topicNames);
Expand Down Expand Up @@ -494,16 +511,51 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo
if (SubscriptionPollRequestType.isValidatedRequestType(requestType)) {
switch (SubscriptionPollRequestType.valueOf(requestType)) {
case POLL:
final Set<String> pollTopicNames = ((PollPayload) request.getPayload()).getTopicNames();
final Set<String> subscribedTopicNames =
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId());
final Set<String> topicNamesToCheck = new HashSet<>(pollTopicNames);
topicNamesToCheck.removeIf(topicName -> !subscribedTopicNames.contains(topicName));
final TSStatus readPermissionStatus =
SubscriptionAgent.topic()
.checkTopicReadPermissions(authenticatedUsername, topicNamesToCheck);
if (readPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return PipeSubscribePollResp.toTPipeSubscribeResp(
readPermissionStatus, Collections.emptyList());
}
events =
handlePipeSubscribePollRequest(
consumerConfig, (PollPayload) request.getPayload(), maxBytes);
break;
case POLL_FILE:
final String tsFileTopicName =
((PollFilePayload) request.getPayload()).getCommitContext().getTopicName();
final TSStatus tsFileReadPermissionStatus =
SubscriptionAgent.topic()
.checkTopicReadPermissions(
authenticatedUsername, Collections.singleton(tsFileTopicName));
if (tsFileReadPermissionStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return PipeSubscribePollResp.toTPipeSubscribeResp(
tsFileReadPermissionStatus, Collections.emptyList());
}
events =
handlePipeSubscribePollTsFileRequest(
consumerConfig, (PollFilePayload) request.getPayload());
break;
case POLL_TABLETS:
final String tabletsTopicName =
((PollTabletsPayload) request.getPayload()).getCommitContext().getTopicName();
final TSStatus tabletsReadPermissionStatus =
SubscriptionAgent.topic()
.checkTopicReadPermissions(
authenticatedUsername, Collections.singleton(tabletsTopicName));
if (tabletsReadPermissionStatus.getCode()
!= TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return PipeSubscribePollResp.toTPipeSubscribeResp(
tabletsReadPermissionStatus, Collections.emptyList());
}
events =
handlePipeSubscribePollTabletsRequest(
consumerConfig, (PollTabletsPayload) request.getPayload());
Expand Down
Loading