Skip to content

fix(cdc): repair default internal principal without relaxing authorization #29635

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

Merged
XuPeng-SH merged 1 commit into from
Oct 5, 2026
Merged

fix(cdc): repair default internal principal without relaxing authorization #29635

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
26 changes: 23 additions & 3 deletions pkg/frontend/internal_executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ func (ie *internalExecutor) ExecWithStatus(ctx context.Context, sql string, opts
ses: sess,
}
defer tempExecCtx.Close()
err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql})
err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql, internalSQLSource: ie.usesDefaultSystemPrincipal(sess, opts)})
res := ie.proto.swapOutResult()
status.AffectedRows = res.affectedRows
if err != nil {
Expand Down Expand Up @@ -240,7 +240,7 @@ func (ie *internalExecutor) Query(ctx context.Context, sql string, opts ie.Sessi
ses: sess,
}
defer tempExecCtx.Close()
err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql})
err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql, internalSQLSource: ie.usesDefaultSystemPrincipal(sess, opts)})
res := ie.proto.swapOutResult()
res.err = moerr.AttachCause(ctx, err)
return res
Expand Down Expand Up @@ -294,7 +294,20 @@ func (ie *internalExecutor) newCmdSession(ctx context.Context, opts ie.SessionOv
now, _ := runtime.ServiceRuntime(ie.service).Clock().Now()
sess.lastCommitTS = now

if ie.baseSessOpts.Username != nil || opts.Username != nil {
if ie.usesDefaultSystemPrincipal(sess, opts) {
// The unnamed system administrator uses reserved user ID 0, whose catalog
// name is root, not the synthetic SQL-source name "internal". Bind
// after both override layers without changing explicit definers or
// bypassing current-principal/role authorization. Other selected roles
// keep their existing fail-closed identity rather than borrowing moadmin.
t.mu.Lock()
t.Tenant = sysAccountName
t.User = rootName
if t.DefaultRoleID == moAdminRoleID {
t.DefaultRole = moAdminRoleName
}
t.mu.Unlock()
} else if ie.baseSessOpts.Username != nil || opts.Username != nil {
if err := bindInternalSessionPrincipal(ctx, sess); err != nil {
sess.Close()
return nil, err
Expand All @@ -305,6 +318,13 @@ func (ie *internalExecutor) newCmdSession(ctx context.Context, opts ie.SessionOv
return sess, nil
}

func (ie *internalExecutor) usesDefaultSystemPrincipal(sess *Session, opts ie.SessionOverrideOptions) bool {
principal := sess.GetTenantInfo()
return ie.baseSessOpts.Username == nil && opts.Username == nil &&
principal.GetTenantID() == sysAccountID && principal.GetUserID() == rootID &&
principal.GetDefaultRoleID() == moAdminRoleID
}

func (ie *internalExecutor) ApplySessionOverride(opts ie.SessionOverrideOptions) {
ie.baseSessOpts = opts
}
Expand Down
212 changes: 212 additions & 0 deletions pkg/frontend/internal_executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,14 @@ package frontend

import (
"context"
"errors"
"fmt"
"testing"

"github.com/prashantv/gostub"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/tidwall/btree"

"github.com/matrixorigin/matrixone/pkg/common/moerr"
"github.com/matrixorigin/matrixone/pkg/common/mpool"
Expand All @@ -29,6 +32,7 @@ import (
"github.com/matrixorigin/matrixone/pkg/container/batch"
"github.com/matrixorigin/matrixone/pkg/container/vector"
"github.com/matrixorigin/matrixone/pkg/defines"
"github.com/matrixorigin/matrixone/pkg/frontend/constant"
"github.com/matrixorigin/matrixone/pkg/taskservice"
"github.com/matrixorigin/matrixone/pkg/testutil"
ie "github.com/matrixorigin/matrixone/pkg/util/internalExecutor"
Expand Down Expand Up @@ -87,6 +91,143 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
bh.sql2result[getSqlForRoleNameOfRoleId(99)] = newMrsForRestoreStringRows([]string{"role_name"}, nil)
stub := gostub.StubFunc(&NewBackgroundExec, bh)
defer stub.Reset()
checkSource := func(t *testing.T, sess *Session, defaultSystem, wantInternal bool) {
t.Helper()
input := &UserInput{sql: "/* cloud_nonuser */ select 1; select 2", internalSQLSource: defaultSystem}
require.False(t, input.isInternal(), "source provenance must not become an internal AST input")
require.False(t, sess.GetIsInternal(), "source provenance must not bypass authorization")
require.Equal(t, wantInternal, input.isInternalSQLSource(sess))
fragments, sources, err := schedulingSQLByStatementWithSQLMode(t.Context(), input.sql, "", input.isInternalSQLSource(sess))
require.NoError(t, err)
require.Len(t, fragments, 2)
wantSources := []string{constant.CloudNoUserSql, constant.ExternSql}
plainSource := constant.ExternSql
if wantInternal {
wantSources = []string{constant.InternalSql, constant.InternalSql}
plainSource = constant.InternalSql
}
require.Equal(t, wantSources, sources)
input.setSqlSourceTypes(sess, sources)
require.Equal(t, wantSources, input.getSqlSourceTypes())
plain := &UserInput{sql: "select 1", internalSQLSource: defaultSystem}
plain.genSqlSourceType(sess)
require.Equal(t, []string{plainSource}, plain.getSqlSourceTypes())
}
// The default numeric sys/root/moadmin identity must agree with the username
// checked by authorization. This must not turn internal execution into
// an authorization bypass or replace any explicitly supplied principal.
sysContext := defines.AttachAccount(t.Context(), 0, 0, 0)
tenantContext := defines.AttachAccount(t.Context(), 10, 42, 43)
for _, tc := range []struct {
name string
ctx context.Context
base, command ie.SessionOverrideOptions
user string
accountID, userID, roleID uint32
defaultSystem bool
}{
{name: "default without context identity", ctx: t.Context(), user: "root", defaultSystem: true},
{name: "default sys account only", ctx: defines.AttachAccountId(t.Context(), 0), user: "root", defaultSystem: true},
{name: "default sys zero IDs", ctx: sysContext, user: "root", defaultSystem: true},
{name: "default sys public role stays fail-closed", ctx: defines.AttachAccount(t.Context(), 0, 0, publicRoleID), user: "internal", roleID: publicRoleID},
{name: "default sys selected role stays fail-closed", ctx: defines.AttachAccount(t.Context(), 0, 0, 43), user: "internal", roleID: 43},
{name: "non-sys zero user is not root", ctx: defines.AttachAccount(t.Context(), 10, 0, 43), accountID: 10, roleID: 43},
{name: "sys nonzero user is not root", ctx: defines.AttachAccount(t.Context(), 0, 42, 43), user: "internal", userID: 42, roleID: 43},
{
name: "base changes account", ctx: sysContext,
base: ie.NewOptsBuilder().AccountId(10).DefaultRoleId(43).Finish(),
user: "internal", accountID: 10, roleID: 43,
},
{
name: "command changes account", ctx: sysContext,
command: ie.NewOptsBuilder().AccountId(10).DefaultRoleId(43).Finish(),
user: "internal", accountID: 10, roleID: 43,
},
{
name: "base changes user", ctx: sysContext,
base: ie.NewOptsBuilder().UserId(42).DefaultRoleId(43).Finish(),
user: "internal", userID: 42, roleID: 43,
},
{
name: "command changes user", ctx: sysContext,
command: ie.NewOptsBuilder().UserId(42).DefaultRoleId(43).Finish(),
user: "internal", userID: 42, roleID: 43,
},
{
name: "base restores sys root but preserves non-admin role", ctx: tenantContext,
base: ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
roleID: 43,
},
{
name: "command restores sys root but preserves non-admin role", ctx: tenantContext,
command: ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
roleID: 43,
},
{
name: "base restores full sys root moadmin identity", ctx: tenantContext,
base: ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(0).Finish(),
user: "root", defaultSystem: true,
},
{
name: "explicit base internal stays internal", ctx: sysContext,
base: ie.NewOptsBuilder().Username("internal").DefaultRoleId(43).Finish(),
user: "internal", roleID: 43,
},
{
name: "explicit command internal stays internal", ctx: sysContext,
command: ie.NewOptsBuilder().Username("sys:internal:moadmin").DefaultRoleId(43).Finish(),
user: "internal", roleID: 43,
},
{
name: "command restores root after base numeric identity", ctx: sysContext,
base: ie.NewOptsBuilder().AccountId(10).UserId(42).DefaultRoleId(43).Finish(),
command: ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(0).Finish(),
user: "root", defaultSystem: true,
},
{
name: "command overrides base root IDs", ctx: sysContext,
base: ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(43).Finish(),
command: ie.NewOptsBuilder().AccountId(10).UserId(42).Finish(),
user: "internal", accountID: 10, userID: 42, roleID: 43,
},
{
name: "numeric command retains explicit base name", ctx: tenantContext,
base: ie.NewOptsBuilder().Username("internal").Finish(),
command: ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
user: "internal", roleID: 43,
},
{
name: "command explicit name overrides base root", ctx: sysContext,
base: ie.NewOptsBuilder().Username("root").Finish(),
command: ie.NewOptsBuilder().Username("internal").DefaultRoleId(43).Finish(),
user: "internal", roleID: 43,
},
} {
t.Run(tc.name, func(t *testing.T) {
executor := newIe("")
executor.ApplySessionOverride(tc.base)
sess, err := executor.newCmdSession(tc.ctx, tc.command)
require.NoError(t, err)
defer sess.Close()
principal := sess.GetTenantInfo()
require.Equal(t, tc.user, principal.GetUser())
require.Equal(t, tc.accountID, principal.GetTenantID())
require.Equal(t, tc.userID, principal.GetUserID())
require.Equal(t, tc.roleID, principal.GetDefaultRoleID())
require.False(t, sess.GetIsInternal(), "normalization must not bypass catalog authorization")
defaultSystem := executor.usesDefaultSystemPrincipal(sess, tc.command)
require.Equal(t, tc.defaultSystem, defaultSystem)
if tc.defaultSystem {
require.Equal(t, sysAccountName, principal.GetTenant())
}
checkSource(t, sess, defaultSystem, tc.defaultSystem || tc.user == "internal")
for _, trusted := range []bool{false, true} {
authSQL := getSqlForActiveRolesForAuthorization(principal, trusted)
require.Contains(t, authSQL, fmt.Sprintf("u.user_id = %d", tc.userID))
require.Contains(t, authSQL, "u.user_name = '"+tc.user+"'")
}
})
}
for _, username := range []string{"definer", "tenant:definer", "tenant:definer:writer", "tenant#definer#writer", "tenant%3Adefiner%3Awriter", "tenant:definer:writer?label=value"} {
t.Run(username, func(t *testing.T) {
executor := newIe("")
Expand All @@ -105,6 +246,8 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
require.Equal(t, "tenant", principal.GetTenant())
require.Equal(t, "selected_writer", principal.GetDefaultRole(), "selected numeric role, not login role")
require.Contains(t, getSqlForActiveRolesForAuthorization(principal, false), "u.user_name = 'definer'")
require.False(t, executor.usesDefaultSystemPrincipal(sess, opts))
checkSource(t, sess, false, false)
})
}
t.Run("root zero IDs", func(t *testing.T) {
Expand All @@ -118,7 +261,76 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
require.Zero(t, sess.GetTenantInfo().GetTenantID())
require.Zero(t, sess.GetTenantInfo().GetUserID())
require.Zero(t, sess.GetTenantInfo().GetDefaultRoleID())
require.False(t, executor.usesDefaultSystemPrincipal(sess, opts))
checkSource(t, sess, false, false)
})
for _, tc := range []struct {
name, externalUser string
roleID uint32
present bool
failure error
}{
{name: "default root catalog present", present: true},
{name: "default root catalog missing"},
{name: "default root catalog lookup error", failure: errors.New("catalog lookup failed")},
{name: "default public role cannot normalize into root", roleID: publicRoleID},
{name: "default selected role cannot normalize into root", roleID: 43},
{name: "external root name is not an authorization bypass", externalUser: "root"},
{name: "external internal name is not an authorization bypass", externalUser: "internal"},
} {
t.Run(tc.name, func(t *testing.T) {
var sess *Session
if tc.externalUser == "" {
executor := newIe("")
opts := ie.NewOptsBuilder().Finish()
var err error
sess, err = executor.newCmdSession(defines.AttachAccount(t.Context(), sysAccountID, rootID, tc.roleID), opts)
require.NoError(t, err)
defer sess.Close()
defaultSystem := executor.usesDefaultSystemPrincipal(sess, opts)
require.Equal(t, tc.roleID == moAdminRoleID, defaultSystem)
checkSource(t, sess, defaultSystem, true)
} else {
// An ordinary session has neither the executor's origin nor
// isInternal set, even when its login name is "internal".
sess = &Session{}
sess.SetTenantInfo(&TenantInfo{Tenant: "sys", User: tc.externalUser, DefaultRole: moAdminRoleName})
checkSource(t, sess, false, tc.externalUser == "internal")
}
authBH := &backgroundExecTest{}
authBH.init()
require.Equal(t, tc.roleID, sess.GetTenantInfo().GetDefaultRoleID())
query := getSqlForActiveRolesForAuthorization(sess.GetTenantInfo(), tc.roleID == moAdminRoleID || tc.roleID == publicRoleID)
wantUser := tc.externalUser
if wantUser == "" {
wantUser = "root"
if tc.roleID != moAdminRoleID {
wantUser = "internal"
}
}
require.Contains(t, query, "u.user_id = 0 and u.user_name = '"+wantUser+"'")
var rows [][]interface{}
if tc.present {
rows = [][]interface{}{{int64(0)}}
}
authBH.sql2result[query] = newMrsForRoleIdOfUserId(rows)
if tc.failure != nil {
authBH.sql2err[query] = tc.failure
}
roles := &btree.Set[int64]{}
err := loadActiveRolesForAuthorization(t.Context(), authBH, sess, roles)
switch {
case tc.failure != nil:
require.ErrorIs(t, err, tc.failure)
case !tc.present:
require.ErrorContains(t, err, "authenticated user no longer matches current catalog")
default:
require.NoError(t, err)
}
require.Equal(t, tc.present, roles.Contains(int64(tc.roleID)))
require.Equal(t, []string{query}, authBH.executedSQLs, "default root must still validate current catalog identity")
})
}
for _, username := range []string{"tenant:", "tenant::writer"} {
for _, base := range []bool{false, true} {
t.Run("invalid/"+username+map[bool]string{false: "/command", true: "/base"}[base], func(t *testing.T) {
Expand Down
3 changes: 3 additions & 0 deletions pkg/frontend/mysql_cmd_executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1396,6 +1396,9 @@ func executeAnalyzeDerivedQuery(ses *Session, outerExecCtx *ExecCtx, sql string)
rewritePolicy: policy,
rewritePolicyMaterialized: true,
}
if outerExecCtx.input != nil {
derivedInput.internalSQLSource = outerExecCtx.input.internalSQLSource
}
if err := doComQuery(ses, &tempExecCtx, derivedInput); err != nil {
return nil, err
}
Expand Down
39 changes: 37 additions & 2 deletions pkg/frontend/mysql_cmd_executor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2614,6 +2614,29 @@ func TestSQLModeStagingDefersRewriteWithRequestSnapshot(t *testing.T) {
bindSource(third)
require.Equal(t, constant.CloudNoUserSql, third.getSqlSourceType(0))
require.Nil(t, input.sqlSourceType, "the unexecuted remainder must not be classified under the old mode")
for _, internalSource := range []bool{false, true} {
t.Run(fmt.Sprintf("source provenance %v", internalSource), func(t *testing.T) {
request := &UserInput{sql: "select * from src.t", rewritePolicy: policy, internalSQLSource: internalSource}
statement := newSQLStatementInput(request, request.sql)
require.NotSame(t, request, statement)
rewritten, err := rewriteSQLStatementInput(ctx, ses, statement)
require.NoError(t, err)
require.NotSame(t, statement, rewritten)
assertMaterializedRemap(t, ctx, rewritten.sql, map[string]string{"src": "dst"})
want := constant.ExternSql
if internalSource {
want = constant.InternalSql
}
for _, copied := range []*UserInput{statement, rewritten} {
require.Equal(t, internalSource, copied.internalSQLSource)
require.False(t, copied.isInternal(), "source propagation must not enable internal-input authorization")
copied.genSqlSourceType(ses)
require.Equal(t, []string{want}, copied.getSqlSourceTypes())
}
require.False(t, ses.GetIsInternal())
require.Nil(t, request.sqlSourceType, "derived source classification must not mutate the request")
})
}
}

func assertMaterializedRemap(t *testing.T, ctx context.Context, sql string, want map[string]string) {
Expand Down Expand Up @@ -3459,19 +3482,30 @@ func TestExecuteAnalyzeDerivedQueryRestoresResponderOnSuccess(t *testing.T) {
defer ctrl.Finish()

ses := newTestSession(t, ctrl)
defer ses.Close()
writer := &countingMysqlWriter{testMysqlWriter: &testMysqlWriter{}}
live := NewMysqlResp(writer)
ses.ReplaceResponser(live)
ctx := defines.AttachAccountId(context.Background(), sysAccountID)
outerExecCtx := &ExecCtx{reqCtx: ctx, ses: ses}
outerExecCtx := &ExecCtx{reqCtx: ctx, ses: ses, input: &UserInput{internalSQLSource: true}}
eng := mock_frontend.NewMockEngine(ctrl)
eng.EXPECT().Hints().Return(engine.Hints{CommitOrRollbackTimeout: time.Second}).AnyTimes()
txnOperator := mock_frontend.NewMockTxnOperator(ctrl)
txnOperator.EXPECT().Txn().Return(txn.TxnMeta{}).AnyTimes()
txnOperator.EXPECT().GetWorkspace().Return(newTestWorkspace()).AnyTimes()
txnOperator.EXPECT().SetFootPrints(gomock.Any(), gomock.Any()).AnyTimes()
txnOperator.EXPECT().Status().Return(txn.TxnStatus_Active).AnyTimes()
txnOperator.EXPECT().TryEnterRunSqlWithTokenAndSQL(gomock.Any(), gomock.Any()).Return(uint64(0), nil).AnyTimes()
observedDerivedSource := false
txnOperator.EXPECT().TryEnterRunSqlWithTokenAndSQL(gomock.Any(), gomock.Any()).DoAndReturn(func(context.CancelFunc, string) (uint64, error) {
derived := ses.GetTxnCompileCtx().execCtx.input
require.NotNil(t, derived)
require.True(t, derived.internalSQLSource)
require.False(t, derived.isInternal())
require.False(t, ses.GetIsInternal())
require.Equal(t, constant.InternalSql, derived.getSqlSourceType(0))
observedDerivedSource = true
return 0, nil
}).AnyTimes()
txnOperator.EXPECT().ExitRunSqlWithToken(gomock.Any()).AnyTimes()
txnOperator.EXPECT().NextSequence().Return(uint64(0)).AnyTimes()
txnOperator.EXPECT().Commit(gomock.Any()).Return(nil).AnyTimes()
Expand All @@ -3482,6 +3516,7 @@ func TestExecuteAnalyzeDerivedQueryRestoresResponderOnSuccess(t *testing.T) {

result, err := executeAnalyzeDerivedQuery(ses, outerExecCtx, "select 1")
require.NoError(t, err)
require.True(t, observedDerivedSource, "derived execution must exercise the provenance assertion")
require.NotNil(t, result)
require.Equal(t, uint64(1), result.GetColumnCount())
require.Equal(t, uint64(1), result.GetRowCount())
Expand Down
Loading