""" 第 15 章 · 权限、配额与多租户 · 实操 Demo ======================================================== 本脚本演示： 1) 用 default 用户准备：库 / 表 / Profile / Role / Quota / Row Policy / Users 2) 用 admin 视角看到全表 3) 用 tenant_acme 身份连接，验证「行级 + 列级」隔离 4) 用 tenant_globex 身份连接，验证看到的是另一份数据 5) 用 tenant_acme 跑批量查询，触发 Quota，看到拒绝错误 6) 清理（可选） 运行： pip install clickhouse-connect python auth_play.py 环境变量（可选）： CK_HOST 默认 127.0.0.1 CK_PORT 默认 8123 CK_USER 默认 default CK_PASS 默认 "" """ import os import sys import time from contextlib import contextmanager try: import clickhouse_connect from clickhouse_connect.driver.exceptions import DatabaseError, OperationalError except ImportError: sys.exit("[ERROR] 需要先安装 clickhouse-connect：pip install clickhouse-connect") CK_HOST = os.getenv("CK_HOST", "127.0.0.1") CK_PORT = int(os.getenv("CK_PORT", "8123")) ADMIN_USER = os.getenv("CK_USER", "default") ADMIN_PASS = os.getenv("CK_PASS", "") DB = "learn_ck" # 演示用户与口令（与 init.sql 保持一致） USERS = { "alice": "alice_pass_2026", "bob": "bob_pass_2026", "tenant_acme": "acme_pass_2026", "tenant_globex": "globex_pass_2026", } # ---------- 工具 ---------- def banner(title: str, ch: str = "─"): print() print(ch * 78) print(f" {title}") print(ch * 78) def connect(user: str, password: str = "", database: str | None = DB): return clickhouse_connect.get_client( host=CK_HOST, port=CK_PORT, username=user, password=password, database=database, connect_timeout=5, send_receive_timeout=10, ) @contextmanager def safe_session(user: str, password: str = "", database: str | None = DB): client = None try: client = connect(user, password, database) yield client finally: if client is not None: client.close() def run_quietly(client, sql: str): """执行单条 SQL，吃掉 IF EXISTS 之类的小报错""" try: client.command(sql) except (DatabaseError, OperationalError) as e: print(f" [warn] {sql[:60]}... -> {type(e).__name__}: {e}") # ---------- Step 1: 准备元数据 ---------- def setup_admin(): banner("Step 1: 用 admin 准备库 / 表 / 角色 / 用户 / 行策略 / 配额") with safe_session(ADMIN_USER, ADMIN_PASS, database=None) as ck: ck.command(f"CREATE DATABASE IF NOT EXISTS {DB}") ck.command(f"DROP TABLE IF EXISTS {DB}.events") ck.command(f""" CREATE TABLE {DB}.events ( event_time DateTime, tenant_id LowCardinality(String), user_id UInt64, event_type LowCardinality(String), properties String, amount Decimal(12, 2) DEFAULT 0 ) ENGINE = MergeTree ORDER BY (tenant_id, event_time, user_id) """) ck.insert( f"{DB}.events", data=[ ("2026-04-17 09:00:00", "acme", 1001, "view", '{"src":"app"}', 0), ("2026-04-17 09:01:00", "acme", 1002, "click", '{"id":"b01"}', 0), ("2026-04-17 09:02:00", "globex", 2001, "view", '{"src":"web"}', 0), ("2026-04-17 09:03:00", "acme", 1001, "add_cart", '{"sku":"A001"}', 0), ("2026-04-17 09:04:00", "globex", 2002, "pay", '{"order":"ord_001"}', 99.00), ("2026-04-17 09:05:00", "initech", 3001, "view", '{"src":"app"}', 0), ("2026-04-17 09:06:00", "acme", 1003, "view", '{"src":"app"}', 0), ("2026-04-17 09:07:00", "globex", 2001, "click", '{"id":"b02"}', 0), ("2026-04-17 09:08:00", "initech", 3002, "pay", '{"order":"ord_002"}', 150.00), ], column_names=[ "event_time", "tenant_id", "user_id", "event_type", "properties", "amount", ], ) # Profile for sql in [ "DROP SETTINGS PROFILE IF EXISTS saas_safe", "DROP SETTINGS PROFILE IF EXISTS dev_full", ]: run_quietly(ck, sql) ck.command(""" CREATE SETTINGS PROFILE saas_safe SETTINGS readonly = 1, max_memory_usage = 5000000000, max_execution_time = 30, max_rows_to_read = 100000000, max_threads = 4, max_concurrent_queries_for_user = 5 """) ck.command(""" CREATE SETTINGS PROFILE dev_full SETTINGS readonly = 0, allow_ddl = 1, max_memory_usage = 20000000000, max_execution_time = 600 """) # Roles for sql in [ "DROP ROLE IF EXISTS analytics_reader", "DROP ROLE IF EXISTS data_engineer", "DROP ROLE IF EXISTS saas_tenant", ]: run_quietly(ck, sql) ck.command("CREATE ROLE analytics_reader") ck.command("CREATE ROLE data_engineer") ck.command("CREATE ROLE saas_tenant") ck.command(f"GRANT SELECT ON {DB}.* TO analytics_reader") ck.command(f"GRANT SHOW TABLES, SHOW DICTIONARIES, SHOW COLUMNS ON {DB}.* TO analytics_reader") ck.command(f""" GRANT SELECT, INSERT, ALTER UPDATE, ALTER DELETE, OPTIMIZE, TRUNCATE, CREATE TABLE, DROP TABLE, ALTER TABLE ON {DB}.* TO data_engineer """) # 列级 GRANT：故意不给 properties，等价于「列级隐藏」 ck.command(f""" GRANT SELECT(event_time, tenant_id, user_id, event_type, amount) ON {DB}.events TO saas_tenant """) ck.command("ALTER ROLE saas_tenant SETTINGS PROFILE 'saas_safe'") # Row Policy run_quietly(ck, f"DROP ROW POLICY IF EXISTS tenant_iso ON {DB}.events") run_quietly(ck, f"DROP ROW POLICY IF EXISTS admin_all ON {DB}.events") ck.command(f""" CREATE ROW POLICY tenant_iso ON {DB}.events FOR SELECT USING tenant_id = substring(currentUser(), length('tenant_') + 1) TO saas_tenant """) ck.command(f""" CREATE ROW POLICY admin_all ON {DB}.events FOR SELECT USING 1 TO data_engineer, analytics_reader """) # Quota：故意把 acme 的小时上限调小，方便后面触发 run_quietly(ck, "DROP QUOTA IF EXISTS saas_quota") ck.command(""" CREATE QUOTA saas_quota KEYED BY user_name FOR INTERVAL 1 HOUR MAX queries = 20, errors = 50, read_rows = 100000, execution_time = 60 TO saas_tenant """) # Users for u in USERS: run_quietly(ck, f"DROP USER IF EXISTS {u}") ck.command(f""" CREATE USER alice IDENTIFIED WITH sha256_password BY '{USERS["alice"]}' DEFAULT ROLE analytics_reader """) ck.command(f""" CREATE USER bob IDENTIFIED WITH sha256_password BY '{USERS["bob"]}' DEFAULT ROLE data_engineer SETTINGS PROFILE 'dev_full' """) ck.command(f""" CREATE USER tenant_acme IDENTIFIED WITH sha256_password BY '{USERS["tenant_acme"]}' DEFAULT ROLE saas_tenant """) ck.command(f""" CREATE USER tenant_globex IDENTIFIED WITH sha256_password BY '{USERS["tenant_globex"]}' DEFAULT ROLE saas_tenant """) ck.command("GRANT analytics_reader TO alice") ck.command("GRANT data_engineer TO bob") ck.command("GRANT saas_tenant TO tenant_acme, tenant_globex") print(" ✓ 元数据准备完成") print(f" ✓ 表 {DB}.events 共 9 行（acme 4, globex 3, initech 2）") # ---------- Step 2: admin 视角 ---------- def view_as_admin(): banner("Step 2: admin 视角 — 应该看到全部 9 行") with safe_session(ADMIN_USER, ADMIN_PASS) as ck: rows = ck.query( f"SELECT tenant_id, count() AS cnt FROM {DB}.events GROUP BY tenant_id ORDER BY tenant_id" ).result_rows for tenant, cnt in rows: print(f" {tenant:<10s} -> {cnt} 行") # ---------- Step 3: tenant_acme 视角 ---------- def view_as_tenant(user: str, expected_tenant: str): banner(f"Step 3/4: 以 {user} 身份连接 — 期望只看到 tenant_id = '{expected_tenant}' 的行") try: with safe_session(user, USERS[user]) as ck: rows = ck.query( f"SELECT tenant_id, count() AS cnt FROM {DB}.events GROUP BY tenant_id" ).result_rows print(" -- 行级隔离效果（GROUP BY tenant_id）：") if not rows: print(" （没有任何行可见）") for tenant, cnt in rows: marker = " ✓" if tenant == expected_tenant else " ✗ 不应出现！" print(f" {tenant:<10s} -> {cnt} 行 {marker}") print(" -- 试访问被列级 GRANT 拦掉的 properties 列：") try: ck.query(f"SELECT properties FROM {DB}.events LIMIT 1") print(" ✗ 居然查到了 — 列级权限失效") except DatabaseError as e: print(f" ✓ 被拒绝：{str(e).splitlines()[0]}") print(" -- 试写入：") try: ck.command( f"INSERT INTO {DB}.events VALUES (now(), '{expected_tenant}', 99, 'hack', '', 0)" ) print(" ✗ 居然写成功了 — readonly 失效") except DatabaseError as e: print(f" ✓ 被拒绝：{str(e).splitlines()[0]}") except OperationalError as e: print(f" [skip] 连接 {user} 失败：{e}") # ---------- Step 5: 触发 Quota ---------- def trigger_quota(user: str = "tenant_acme"): banner(f"Step 5: 用 {user} 连续跑查询，触发 Quota（per-hour queries = 20）") with safe_session(user, USERS[user]) as ck: ok, denied = 0, 0 for i in range(35): try: ck.query(f"SELECT count() FROM {DB}.events") ok += 1 except DatabaseError as e: denied += 1 msg = str(e).splitlines()[0] if "Quota" in msg or "quota" in msg or "QUOTA_EXCEEDED" in msg: print(f" 第 {i+1} 次：⛔ {msg}") else: print(f" 第 {i+1} 次：[err] {msg}") if denied >= 3: break time.sleep(0.05) print(f"\n 统计：成功 {ok} 次，被拒 {denied} 次") banner("配额实时使用情况（system.quotas_usage）") with safe_session(ADMIN_USER, ADMIN_PASS) as ck: rows = ck.query(""" SELECT quota_name, quota_key, duration, queries, max_queries, read_rows, max_read_rows FROM system.quotas_usage WHERE quota_name = 'saas_quota' ORDER BY duration """).result_rows for r in rows: print( f" quota={r[0]} key={r[1]} dur={r[2]}s " f"queries={r[3]}/{r[4]} read_rows={r[5]}/{r[6]}" ) # ---------- Step 6: 清理 ---------- def cleanup(): banner("Step 6: 清理（DROP USER / ROLE / QUOTA / POLICY / PROFILE）", "·") with safe_session(ADMIN_USER, ADMIN_PASS, database=None) as ck: for sql in [ "DROP USER IF EXISTS alice, bob, tenant_acme, tenant_globex", "DROP ROLE IF EXISTS analytics_reader, data_engineer, saas_tenant", "DROP QUOTA IF EXISTS saas_quota", f"DROP ROW POLICY IF EXISTS tenant_iso ON {DB}.events", f"DROP ROW POLICY IF EXISTS admin_all ON {DB}.events", "DROP SETTINGS PROFILE IF EXISTS saas_safe", "DROP SETTINGS PROFILE IF EXISTS dev_full", ]: run_quietly(ck, sql) print(" ✓ 清理完成") # ---------- main ---------- def main(): print(f"==> ClickHouse @ {CK_HOST}:{CK_PORT}, admin = {ADMIN_USER}") try: setup_admin() except (DatabaseError, OperationalError) as e: sys.exit( f"\n[FATAL] 初始化失败：{e}\n" " 请确保 default 用户具备 access_management = 1，或换一个有该权限的账号。" ) view_as_admin() view_as_tenant("tenant_acme", "acme") view_as_tenant("tenant_globex", "globex") trigger_quota("tenant_acme") if os.getenv("CK_KEEP", "0") != "1": cleanup() else: print("\n 注：检测到 CK_KEEP=1，跳过清理，保留对象供继续探索。") if __name__ == "__main__": main()