This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6a07201308e [fix](load) Enforce auth for group commit stream load
(#69516)
6a07201308e is described below
commit 6a07201308ed4bc450716e0748e68c70aeb82b86
Author: Gavin Chou <[email protected]>
AuthorDate: Mon Oct 26 11:05:45 2026 +0800
[fix](load) Enforce auth for group commit stream load (#69516)
Problem Summary:
Explicit group commit Stream Load skips the normal transaction begin
path. FE planning therefore needs to apply the same credential and
target-table LOAD checks before follower forwarding or local planning.
### Release note
Enforce authentication and target-table LOAD authorization for explicit
group commit Stream Load.
---
.../apache/doris/service/FrontendServiceImpl.java | 37 ++++++++++----
.../doris/service/FrontendServiceImplTest.java | 56 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 50873fb01d6..5078dc05228 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -2983,6 +2983,26 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
return Math.floorMod(GROUP_COMMIT_FOLLOWER_INDEX.getAndIncrement(),
followerCount);
}
+ private void checkGroupCommitStreamLoadAuth(TStreamLoadPutRequest request)
throws AuthenticationException {
+ // Explicit group commit skips loadTxnBegin, so authenticate and
authorize the standard Stream Load
+ // request here before it can be forwarded or planned. Requests with
load_sql use the separate HTTP
+ // Stream path, whose planner performs its own authorization and whose
internal caller uses a token.
+ if (!Strings.isNullOrEmpty(request.getLoadSql()) ||
!request.isSetGroupCommitMode()
+ || request.getGroupCommitMode().equalsIgnoreCase("off_mode")) {
+ return;
+ }
+
+ if (request.isSetAuthCode()) {
+ // TODO: deprecated, removed in 3.1, use token instead.
+ } else if (Strings.isNullOrEmpty(request.getToken())) {
+ checkSingleTablePasswordAndPrivs(request.getUser(),
request.getPasswd(), request.getDb(),
+ request.getTbl(), request.getUserIp(), PrivPredicate.LOAD,
+ toForwardedCertificateInfo(request.getCertBasedAuth()));
+ } else {
+ checkTokenOrThrow(request.getToken());
+ }
+ }
+
private TStreamLoadPutResult
forwardGroupCommitStreamLoad(TStreamLoadPutRequest request) {
HostInfo selfNode = Env.getCurrentEnv().getSelfNode();
List<Frontend> followers =
Env.getCurrentEnv().getFrontends(FrontendNodeType.FOLLOWER).stream()
@@ -3052,14 +3072,6 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
}
String groupCommitMode = request.getGroupCommitMode();
- if (groupCommitMode != null && !groupCommitMode.equals("off_mode") &&
Env.getCurrentEnv().isMaster()
- && Config.enable_forward_group_commit_stream_load_to_follower)
{
- TStreamLoadPutResult result =
forwardGroupCommitStreamLoad(request);
- if (result != null) {
- return result;
- }
- }
-
TStreamLoadPutResult result = new TStreamLoadPutResult();
TStatus status = new TStatus(TStatusCode.OK);
result.setStatus(status);
@@ -3067,6 +3079,15 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
StreamLoadHandler streamLoadHandler = new StreamLoadHandler(request,
null, result, clientAddr);
try {
+ checkGroupCommitStreamLoadAuth(request);
+ if (groupCommitMode != null && !groupCommitMode.equals("off_mode")
&& Env.getCurrentEnv().isMaster()
+ &&
Config.enable_forward_group_commit_stream_load_to_follower) {
+ TStreamLoadPutResult forwardedResult =
forwardGroupCommitStreamLoad(request);
+ if (forwardedResult != null) {
+ return forwardedResult;
+ }
+ }
+
streamLoadHandler.setCloudCluster();
List<TPipelineWorkloadGroup> tWorkloadGroupList = null;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
index efa76702539..6c139546424 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
@@ -68,8 +68,11 @@ import org.apache.doris.thrift.TShowProcessListRequest;
import org.apache.doris.thrift.TShowUserRequest;
import org.apache.doris.thrift.TStatus;
import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TStreamLoadPutRequest;
+import org.apache.doris.thrift.TStreamLoadPutResult;
import org.apache.doris.thrift.TTableStatus;
import org.apache.doris.thrift.TTabletLocation;
+import org.apache.doris.thrift.TUniqueId;
import org.apache.doris.transaction.GlobalTransactionMgrIface;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.WriteBlockAllocatingTransaction;
@@ -1159,6 +1162,59 @@ public class FrontendServiceImplTest extends
TestWithFeService {
}
}
+ @Test
+ public void
testGroupCommitStreamLoadPutChecksAuthenticationAndLoadPrivilege() throws
Exception {
+ String noLoadUser = "group_commit_no_load_user";
+ String noLoadPassword = "group_commit_test_password";
+ boolean originalForwardSetting =
Config.enable_forward_group_commit_stream_load_to_follower;
+ Config.enable_forward_group_commit_stream_load_to_follower = false;
+ executeCommand("DROP USER IF EXISTS " + noLoadUser);
+ executeCommand("CREATE USER " + noLoadUser + " IDENTIFIED BY '" +
noLoadPassword + "'");
+
+ try {
+ FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+
+ TStreamLoadPutResult badPassword = impl.streamLoadPut(
+ groupCommitStreamLoadRequest("root", "wrong_password",
"missing_table"));
+ Assertions.assertEquals(TStatusCode.ANALYSIS_ERROR,
badPassword.getStatus().getStatusCode());
+
Assertions.assertTrue(badPassword.getStatus().getErrorMsgs().get(0).contains("Access
denied"));
+
+ TStreamLoadPutResult missingLoad = impl.streamLoadPut(
+ groupCommitStreamLoadRequest(noLoadUser, noLoadPassword,
"missing_table"));
+ Assertions.assertEquals(TStatusCode.ANALYSIS_ERROR,
missingLoad.getStatus().getStatusCode());
+ String missingLoadError =
missingLoad.getStatus().getErrorMsgs().get(0);
+ Assertions.assertTrue(missingLoadError.contains("Load_priv"),
missingLoadError);
+
+ FrontendServiceImpl tokenCheckingImpl = Mockito.spy(new
FrontendServiceImpl(exeEnv));
+ TStreamLoadPutRequest badTokenRequest =
groupCommitStreamLoadRequest("", "", "missing_table");
+ badTokenRequest.setToken("bad-token");
+
Mockito.doReturn(false).when(tokenCheckingImpl).checkToken("bad-token");
+ TStreamLoadPutResult badToken =
tokenCheckingImpl.streamLoadPut(badTokenRequest);
+ Assertions.assertEquals(TStatusCode.ANALYSIS_ERROR,
badToken.getStatus().getStatusCode());
+
Assertions.assertTrue(badToken.getStatus().getErrorMsgs().get(0).contains("Invalid
token"));
+
+ TStreamLoadPutResult authorized = impl.streamLoadPut(
+ groupCommitStreamLoadRequest("root", "", "missing_table"));
+ Assertions.assertEquals(TStatusCode.NOT_FOUND,
authorized.getStatus().getStatusCode());
+
Assertions.assertTrue(authorized.getStatus().getErrorMsgs().get(0).contains("missing_table"));
+ } finally {
+ connectContext.setThreadLocalInfo();
+ executeCommand("DROP USER IF EXISTS " + noLoadUser);
+ Config.enable_forward_group_commit_stream_load_to_follower =
originalForwardSetting;
+ }
+ }
+
+ private static TStreamLoadPutRequest groupCommitStreamLoadRequest(String
user, String password, String table) {
+ return new TStreamLoadPutRequest()
+ .setUser(user)
+ .setPasswd(password)
+ .setDb("test")
+ .setTbl(table)
+ .setUserIp("127.0.0.1")
+ .setLoadId(new TUniqueId(1L, 1L))
+ .setGroupCommitMode("async_mode");
+ }
+
private MockedStatic<Env> transactionValidationEnvMock;
private void mockTransactionForTokenValidation(long txnId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
