|
// Reproducer for a TOCTOU race in ClickHouse ReplicatedAccessStorage. |
|
// |
|
// When RBAC is stored in Keeper (<user_directories><replicated>), creating a |
|
// ROLE and then immediately referencing it by name in CREATE QUOTA ... TO <role> |
|
// (or CREATE ROW POLICY ... TO <role>) can fail with: |
|
// |
|
// Code: 511. There is no role `...` in `user directories`. (UNKNOWN_ROLE) |
|
// |
|
// even though CREATE ROLE just succeeded over the same session and CREATE USER |
|
// ... DEFAULT ROLE resolved the same role. |
|
// |
|
// Root cause (src/Access/ZooKeeperReplicator.cpp, refreshEntities(all=false)): |
|
// the background watch thread reads getChildrenWatch("/uuid") WITHOUT the mutex, |
|
// then later takes the mutex and calls memory_storage.removeAllExcept(snapshot). |
|
// If the snapshot was taken before the freshly inserted role's znode existed, |
|
// the role is evicted from the local in-memory cache. |
|
// |
|
// This mirrors clickhouse-backup TestRBAC -> createRBACObjects. To widen the |
|
// narrow TOCTOU window we (1) bloat the /uuid list with many entities so the |
|
// lock-free getChildrenWatch + parse takes longer, and (2) run dedicated churn |
|
// goroutines so a stale-snapshot refresh is almost always in flight. |
|
package main |
|
|
|
import ( |
|
"context" |
|
"fmt" |
|
"os" |
|
"strconv" |
|
"strings" |
|
"sync" |
|
"sync/atomic" |
|
"time" |
|
|
|
"github.com/ClickHouse/clickhouse-go/v2" |
|
) |
|
|
|
func envInt(name string, def int) int { |
|
if v := os.Getenv(name); v != "" { |
|
if n, err := strconv.Atoi(v); err == nil { |
|
return n |
|
} |
|
} |
|
return def |
|
} |
|
|
|
func addr() string { |
|
if a := os.Getenv("CLICKHOUSE_ADDR"); a != "" { |
|
return a |
|
} |
|
return "127.0.0.1:9000" |
|
} |
|
|
|
func open() (clickhouse.Conn, error) { |
|
return clickhouse.Open(&clickhouse.Options{ |
|
Addr: []string{addr()}, |
|
Auth: clickhouse.Auth{Database: "default", Username: "default"}, |
|
MaxOpenConns: 1, |
|
MaxIdleConns: 0, |
|
DialTimeout: 10 * time.Second, |
|
}) |
|
} |
|
|
|
func isUnknownRole(err error) bool { |
|
if err == nil { |
|
return false |
|
} |
|
s := err.Error() |
|
return strings.Contains(s, "UNKNOWN_ROLE") || |
|
strings.Contains(s, "code: 511") || |
|
strings.Contains(s, "There is no role") |
|
} |
|
|
|
var ( |
|
iters atomic.Int64 |
|
hit atomic.Bool |
|
once sync.Once |
|
stop = make(chan struct{}) |
|
) |
|
|
|
func stopped() bool { |
|
select { |
|
case <-stop: |
|
return true |
|
default: |
|
return false |
|
} |
|
} |
|
|
|
// victim mirrors test/integration/rbac_test.go:createRBACObjects: create a |
|
// SETTINGS PROFILE, ROLE, USER, then reference the role by name in CREATE QUOTA |
|
// / CREATE ROW POLICY (TO <role>), dropping first to recreate each iteration. |
|
func victim(ctx context.Context, conn clickhouse.Conn, name string) (failedStmt string, err error) { |
|
q := func(query string) bool { |
|
if err = conn.Exec(ctx, query); err != nil { |
|
failedStmt = query |
|
return false |
|
} |
|
return true |
|
} |
|
for _, drop := range []string{ |
|
fmt.Sprintf("DROP SETTINGS PROFILE IF EXISTS `%s`", name), |
|
fmt.Sprintf("DROP QUOTA IF EXISTS `%s`", name), |
|
fmt.Sprintf("DROP ROW POLICY IF EXISTS `%s` ON test_rbac.test_rbac", name), |
|
fmt.Sprintf("DROP ROLE IF EXISTS `%s`", name), |
|
fmt.Sprintf("DROP USER IF EXISTS `%s`", name), |
|
} { |
|
if !q(drop) { |
|
return |
|
} |
|
} |
|
if !q(fmt.Sprintf("CREATE SETTINGS PROFILE `%s` SETTINGS max_execution_time=60", name)) { |
|
return |
|
} |
|
if !q(fmt.Sprintf("CREATE ROLE `%s` SETTINGS PROFILE `%s`", name, name)) { |
|
return |
|
} |
|
if !q(fmt.Sprintf("CREATE USER `%s` IDENTIFIED BY 'test_rbac_password' DEFAULT ROLE `%s`", name, name)) { |
|
return |
|
} |
|
// The statement that fails with UNKNOWN_ROLE in CI (RolesOrUsersSet::init). |
|
if !q(fmt.Sprintf("CREATE QUOTA `%s` KEYED BY user_name FOR INTERVAL 1 hour NO LIMITS TO `%s`", name, name)) { |
|
return |
|
} |
|
if !q(fmt.Sprintf("CREATE ROW POLICY `%s` ON test_rbac.test_rbac USING v>=0 AS RESTRICTIVE TO `%s`", name, name)) { |
|
return |
|
} |
|
return |
|
} |
|
|
|
func main() { |
|
ctx := context.Background() |
|
seed := envInt("SEED_ENTITIES", 6000) |
|
churners := envInt("CHURNERS", 24) |
|
victims := envInt("VICTIMS", 16) |
|
runSecs := envInt("RUN_SECS", 240) |
|
|
|
setup, err := open() |
|
if err != nil { |
|
panic(err) |
|
} |
|
for _, q := range []string{ |
|
"CREATE DATABASE IF NOT EXISTS test_rbac", |
|
"CREATE TABLE IF NOT EXISTS test_rbac.test_rbac (v UInt64) ENGINE=MergeTree() ORDER BY tuple()", |
|
} { |
|
if err := setup.Exec(ctx, q); err != nil { |
|
panic(err) |
|
} |
|
} |
|
|
|
// Bloat the /uuid children list so the lock-free getChildrenWatch + parse in |
|
// refreshEntities is slow, widening the TOCTOU window. |
|
fmt.Printf("seeding %d dummy settings profiles to bloat /uuid ...\n", seed) |
|
for i := 0; i < seed; i++ { |
|
if err := setup.Exec(ctx, fmt.Sprintf("CREATE SETTINGS PROFILE IF NOT EXISTS `seed-%d` SETTINGS max_execution_time=%d", i, i%100+1)); err != nil { |
|
panic(err) |
|
} |
|
if i%1000 == 999 { |
|
fmt.Printf(" seeded %d\n", i+1) |
|
} |
|
} |
|
_ = setup.Close() |
|
fmt.Println("seed done; starting churn + victims") |
|
|
|
report := func(workerName, stmt string, err error) { |
|
once.Do(func() { |
|
hit.Store(true) |
|
fmt.Println("\n========================================================") |
|
fmt.Println("REPRODUCED: ReplicatedAccessStorage TOCTOU race") |
|
fmt.Println("========================================================") |
|
fmt.Printf("victim entity : %s\n", workerName) |
|
fmt.Printf("total iters : %d\n", iters.Load()) |
|
fmt.Printf("failed stmt : %s\n", stmt) |
|
fmt.Printf("server error : %v\n", err) |
|
fmt.Println("--------------------------------------------------------") |
|
fmt.Println("CREATE ROLE succeeded and CREATE USER ... DEFAULT ROLE resolved") |
|
fmt.Println("the same role, yet the next TO `<role>` failed: the role was") |
|
fmt.Println("evicted from the local cache by the background watch thread.") |
|
close(stop) |
|
}) |
|
} |
|
|
|
var wg sync.WaitGroup |
|
|
|
// Churn goroutines: rapid CREATE/DROP of unique profiles to keep the /uuid |
|
// children list volatile, so a stale-snapshot watch refresh is always in flight. |
|
for c := 0; c < churners; c++ { |
|
wg.Add(1) |
|
go func(id int) { |
|
defer wg.Done() |
|
conn, err := open() |
|
if err != nil { |
|
return |
|
} |
|
defer conn.Close() |
|
n := 0 |
|
for !stopped() { |
|
name := fmt.Sprintf("churn-%d-%d", id, n%4) |
|
_ = conn.Exec(ctx, fmt.Sprintf("CREATE SETTINGS PROFILE IF NOT EXISTS `%s` SETTINGS max_execution_time=%d", name, n%50+1)) |
|
_ = conn.Exec(ctx, fmt.Sprintf("DROP SETTINGS PROFILE IF EXISTS `%s`", name)) |
|
n++ |
|
} |
|
}(c) |
|
} |
|
|
|
// Victim goroutines: the role -> quota sequence that exposes the eviction. |
|
for w := 0; w < victims; w++ { |
|
wg.Add(1) |
|
go func(id int) { |
|
defer wg.Done() |
|
conn, err := open() |
|
if err != nil { |
|
return |
|
} |
|
defer conn.Close() |
|
name := fmt.Sprintf("test.rbac-name-%d", id) |
|
for !stopped() { |
|
stmt, err := victim(ctx, conn, name) |
|
iters.Add(1) |
|
if isUnknownRole(err) { |
|
report(name, stmt, err) |
|
return |
|
} |
|
} |
|
}(w) |
|
} |
|
|
|
// Progress + timeout watchdog. |
|
go func() { |
|
t := time.NewTicker(2 * time.Second) |
|
defer t.Stop() |
|
deadline := time.Now().Add(time.Duration(runSecs) * time.Second) |
|
for { |
|
select { |
|
case <-stop: |
|
return |
|
case <-t.C: |
|
fmt.Printf("... %d victim iterations, no race yet\n", iters.Load()) |
|
if time.Now().After(deadline) { |
|
fmt.Println("timeout: race not reproduced in this run") |
|
once.Do(func() { close(stop) }) |
|
return |
|
} |
|
} |
|
} |
|
}() |
|
|
|
wg.Wait() |
|
if !hit.Load() { |
|
os.Exit(2) |
|
} |
|
} |
