datastore/tests/integration/test_iam_auth/test.py

"""The Hanzo IAM user directory end to end: tokens from a stand-in issuer, the image's own directory
section and bootstrap roles, and the event-plane schema (hanzoai/o11y
deploy/datastore/migrations/0005_iam_roles.sql) applied by the schema's owning application under
its own token, with no password anywhere.
"""

import os
import shlex
import typing
import urllib.parse

import pytest

from helpers.cluster import ClickHouseCluster
from helpers.test_tools import wait_condition

cluster = ClickHouseCluster(__file__)
instance = cluster.add_instance(
    "node",
    # The image's own directory section and bootstrap roles, not copies of them; the overlay
    # points the issuer at iam_server.py and changes nothing else.
    main_configs=["../../../docker/server/datastore-iam.xml", "configs/test-issuer.xml"],
    user_configs=["../../../docker/server/datastore-iam-roles.xml"],
)
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))

# The event plane and its rows, as the o11y application.
PLANE = [
    "CREATE DATABASE IF NOT EXISTS event",
    "CREATE TABLE IF NOT EXISTS event.fact (org LowCardinality(String), id String) ENGINE = MergeTree ORDER BY (org, id)",
    "CREATE TABLE IF NOT EXISTS event.sample (org LowCardinality(String), value Float64) ENGINE = MergeTree ORDER BY org",
    "INSERT INTO event.fact VALUES ('acme', 'a1'), ('acme', 'a2'), ('zoo', 'z1'), ('', 'none')",
]

# The statements of hanzoai/o11y deploy/datastore/migrations/0005_iam_roles.sql, in its order: roles,
# then the policies that confine them, then the grants.
MIGRATION = [
    "CREATE ROLE IF NOT EXISTS tenant",
    "CREATE ROLE IF NOT EXISTS reader",
    "CREATE ROLE IF NOT EXISTS writer",
    "CREATE ROLE IF NOT EXISTS sink",
    "CREATE ROW POLICY OR REPLACE tenant ON event.* USING org = currentOrg() AS PERMISSIVE TO tenant",
    "CREATE ROW POLICY OR REPLACE superadmin ON event.* USING 1 AS PERMISSIVE TO superadmin",
    "CREATE ROW POLICY OR REPLACE schema ON event.* USING 1 AS PERMISSIVE TO schema",
    "GRANT SELECT ON event.* TO reader",
    "GRANT INSERT ON event.* TO writer",
    "REVOKE INSERT ON event.fact FROM writer",
    "GRANT INSERT ON event.fact TO sink",
    """SELECT throwIf(
        (SELECT count() FROM system.roles
          WHERE name IN ('tenant', 'reader', 'writer', 'sink', 'schema', 'platform', 'superadmin')) != 7
     OR (SELECT count() FROM system.row_policies WHERE database = 'event') != 3
     OR (SELECT count() FROM system.row_policies
          WHERE database = 'event' AND table = ''
            AND ((short_name = 'tenant' AND select_filter = 'org = currentOrg()' AND apply_to_list = ['tenant'])
              OR (short_name = 'superadmin' AND select_filter = '1' AND apply_to_list = ['superadmin'])
              OR (short_name = 'schema' AND select_filter = '1' AND apply_to_list = ['schema']))) != 3,
        'The IAM roles or the event.* row policies are not as this file declares them.') AS ok""",
]

# The four event rows every test starts from, by org: acme 2, zoo 1, and one with no org.
FACTS = 4


def token(**claims: str) -> str:
    query = urllib.parse.urlencode(claims)
    return instance.exec_in_container(["curl", "-sf", f"http://localhost:8000/token?{query}"])


def person(sub: str, orgs: str, **claims: str) -> str:
    return token(sub=sub, orgs=orgs, **claims)


def program(application: str, **claims: str) -> str:
    """A client_credentials token of the platform's application row admin/<application>, as IAM mints it."""
    return token(sub=f"admin/{application}", owner="hanzo", azp=application, type="application", **claims)


def confined(bearer: str) -> None:
    """A person of acme reads acme's rows, or is refused; never another org's."""
    code, body = http("SELECT org, count() FROM event.fact GROUP BY org ORDER BY org FORMAT TSV", bearer)
    assert code != "200" or body in ("", "acme\t2\n"), body


def http(sql: str, bearer: str, extra: str = "") -> typing.Tuple[str, str]:
    """(status code, body) of a query over HTTP with `Authorization: Bearer`."""
    command = (
        "curl -s -w '\\n%{http_code}' -H "
        + shlex.quote(f"Authorization: Bearer {bearer}")
        + " --data-binary "
        + shlex.quote(sql)
        + " "
        + shlex.quote(f"http://localhost:8123/{extra}")
    )
    out = instance.exec_in_container(["bash", "-c", command])
    body, _, code = out.rpartition("\n")
    return code, body


def ok(sql: str, bearer: str, extra: str = "") -> str:
    code, body = http(sql, bearer, extra)
    assert code == "200", f"{sql}: {body}"
    return body


def denied(sql: str, bearer: str) -> str:
    code, body = http(sql, bearer)
    assert code != "200", f"{sql} was allowed: {body}"
    return body


def native(sql: str, jwt: str) -> str:
    """Queries over the native protocol, the client logging in with `--jwt`."""
    return instance.exec_in_container(
        ["bash", "-c", f"/usr/bin/clickhouse client --jwt {shlex.quote(jwt)} --query {shlex.quote(sql)}"]
    )


@pytest.fixture(scope="module")
def started_cluster() -> typing.Generator[ClickHouseCluster, None, None]:
    try:
        cluster.start()
        instance.copy_file_to_container(os.path.join(SCRIPT_DIR, "iam_server.py"), "/iam_server.py")
        instance.exec_in_container(
            ["bash", "-c", "python3 /iam_server.py > /var/log/clickhouse-server/iam_server.log 2>&1"],
            detach=True,
            user="root",
        )
        wait_condition(
            lambda: instance.exec_in_container(["curl", "-s", "http://localhost:8000/health"], nothrow=True),
            lambda response: response == "OK",
            max_attempts=40,
            delay=0.5,
        )
        # The host creates the database it owns.
        ok("CREATE DATABASE IF NOT EXISTS hanzo", program("hanzo-cloud"))
        # The schema's owner applies its own migration as its own identity. Wherever the run stops, a
        # person reads their own org's rows or none: never every org's.
        schema = program("hanzo-o11y")
        for statement in PLANE:
            ok(statement, schema)
        for statement in MIGRATION:
            ok(statement, schema)
            confined(person("acme/alice", "acme"))
        yield cluster
    finally:
        cluster.shutdown()


def test_person_sees_only_their_home_org(started_cluster):
    by_org = "SELECT org, count() FROM event.fact GROUP BY org ORDER BY org FORMAT TSV"
    assert ok(by_org, person("acme/alice", "acme")) == "acme\t2\n"
    assert ok(by_org, person("zoo/bob", "zoo,acme")) == "zoo\t1\n"


def test_owner_decides_nothing(started_cluster):
    # Signing in through an application the admin org owns, under the admin org's key: the tenant is
    # still the home org, and it is no SuperAdmin.
    bearer = person("acme/alice", "acme", owner="admin", kid="cert-admin")
    assert ok("SELECT count() FROM event.fact", bearer) == "2\n"
    assert ok("SELECT currentOrg(), has(currentRoles(), 'superadmin') FORMAT TSV", bearer) == "acme\t0\n"


def test_token_without_memberships_reads_nothing(started_cluster):
    # No membership set: no org, and not a person, so not even a reader.
    bearer = token(sub="acme/carol", owner="acme")
    assert ok("SELECT currentOrg() IS NULL, arraySort(currentRoles()) FORMAT TSV", bearer) == "1\t['tenant']\n"
    assert "ACCESS_DENIED" in denied("SELECT count() FROM event.fact", bearer)


def test_superadmin_is_home_admin_under_the_admin_key(started_cluster):
    assert ok("SELECT count() FROM event.fact", person("admin/z", "admin", kid="cert-admin")) == f"{FACTS}\n"
    # The same claims under another org's key are an ordinary admin-org person: no rows but admin's.
    assert ok("SELECT count() FROM event.fact", person("admin/z", "admin", kid="cert-hanzo")) == "0\n"
    # An admin membership next to a brand home org is not SuperAdmin either.
    assert ok("SELECT count() FROM event.fact", person("hanzo/z", "hanzo,admin", kid="cert-admin")) == "0\n"


def test_identity_is_the_token(started_cluster):
    body = ok("SELECT currentUser(), currentOrg(), arraySort(currentRoles()) FORMAT TSV", person("acme/alice", "acme"))
    assert body == "acme/alice\tacme\t['reader','tenant']\n"
    # The ephemeral user can be shown; it has no stored credential to show.
    assert "IDENTIFIED WITH jwt" in ok("SHOW CREATE USER `acme/alice`", person("admin/z", "admin", kid="cert-admin"))


def test_event_app_inserts_facts_and_reads_nothing(started_cluster):
    event = program("hanzo-event")
    assert ok("SELECT arraySort(currentRoles()) FORMAT TSV", event) == "['sink','tenant']\n"
    ok("INSERT INTO event.fact VALUES ('hanzo', 'h1')", event)
    # INSERT on event.fact and nothing else: no read of any org's rows, no other table.
    assert "ACCESS_DENIED" in denied("SELECT count() FROM event.fact", event)
    assert "ACCESS_DENIED" in denied("INSERT INTO event.sample VALUES ('hanzo', 1)", event)
    assert "ACCESS_DENIED" in denied("ALTER TABLE event.fact DELETE WHERE 1", event)
    ok("DELETE FROM event.fact WHERE id = 'h1'", program("hanzo-o11y"))


def test_telemetry_writer_writes_everything_but_the_facts(started_cluster):
    # `writer` is every telemetry table in the plane but event.fact. (The o11y application also
    # holds `schema`, whose migrations may backfill any event table, facts included.)
    superadmin = person("admin/z", "admin", kid="cert-admin")
    grants = ok("SHOW GRANTS FOR writer FORMAT TSV", superadmin)
    assert grants == "GRANT INSERT ON event.* TO writer\nREVOKE INSERT ON event.fact FROM writer\n", grants


def test_no_other_application_is_the_sink(started_cluster):
    # A tenant that registered the client id hanzo-event in its own org first: IAM signs its own row,
    # evilcorp/hanzo-event, as `sub`. A client id alone names nobody.
    squatter = token(sub="evilcorp/hanzo-event", owner="evilcorp", azp="hanzo-event", type="application", orgs="evilcorp")
    assert ok("SELECT arraySort(currentRoles()) FORMAT TSV", squatter) == "['tenant']\n"
    for bearer in [
        squatter,
        program("hanzo-cloud"),
        # The platform's row held by another client (a delegated or exchanged token).
        token(sub="admin/hanzo-event", owner="hanzo", azp="hanzo-sandbox", type="application"),
        # A person signed in through the event app's application is a reader, not the sink.
        person("hanzo/dan", "hanzo", azp="hanzo-event"),
        # A token that calls itself a program in any word but IAM's is no program.
        token(sub="admin/hanzo-event", azp="hanzo-event", type="program"),
    ]:
        assert "ACCESS_DENIED" in denied("INSERT INTO event.fact VALUES ('hanzo', 'x')", bearer)
    # Nor does the squatter read anything, its own org's rows included: it holds no SELECT.
    assert "ACCESS_DENIED" in denied("SELECT count() FROM event.fact", squatter)


def test_platform_owns_the_hanzo_database(started_cluster):
    host = program("hanzo-cloud")
    assert ok("SELECT arraySort(currentRoles()) FORMAT TSV", host) == "['platform','tenant']\n"
    ok("CREATE TABLE IF NOT EXISTS hanzo.cloud_usage (organization String, tokens UInt32) ENGINE = MergeTree ORDER BY organization", host)
    ok("INSERT INTO hanzo.cloud_usage VALUES ('acme', 10), ('zoo', 20)", host)
    ok("ALTER TABLE hanzo.cloud_usage ADD COLUMN IF NOT EXISTS model String", host)
    # The event plane it neither reads nor writes nor reshapes: an org's read is that org's token.
    assert "ACCESS_DENIED" in denied("SELECT count() FROM event.fact", host)
    assert "ACCESS_DENIED" in denied("INSERT INTO event.fact VALUES ('hanzo', 'x')", host)
    assert "ACCESS_DENIED" in denied("CREATE TABLE event.escape (org String) ENGINE = Memory", host)
    assert "ACCESS_DENIED" in denied("CREATE TABLE default.escape (x UInt8) ENGINE = Memory", host)
    assert "ACCESS_DENIED" in denied("CREATE ROLE intruder", host)
    assert "ACCESS_DENIED" in denied("GRANT SELECT ON event.* TO sink", host)

    # It confines its own org-keyed tables for the reads cloud makes as a person's org, the way
    # 0005 confines the event plane: the tenant's rows only, every row for itself, and only then
    # SELECT for readers.
    ok("CREATE ROW POLICY OR REPLACE tenant ON hanzo.cloud_usage USING organization = currentOrg() AS PERMISSIVE TO tenant", host)
    ok("CREATE ROW POLICY OR REPLACE platform ON hanzo.* USING 1 AS PERMISSIVE TO platform", host)
    ok("GRANT SELECT ON hanzo.cloud_usage TO reader", host)
    assert ok("SELECT organization, sum(tokens) FROM hanzo.cloud_usage GROUP BY organization FORMAT TSV", person("acme/alice", "acme")) == "acme\t10\n"
    assert ok("SELECT sum(tokens) FROM hanzo.cloud_usage", host) == "30\n"
    # Nobody else touches the host's database.
    for other in [program("hanzo-event"), program("hanzo-o11y")]:
        assert "ACCESS_DENIED" in denied("SELECT count() FROM hanzo.cloud_usage", other)


def test_a_view_an_identity_defines_holds_no_authority_of_its_own(started_cluster):
    # The token's roles are the session's. A materialized view defined under them gets a stored
    # definer, `<sub>:definer`, with none of them, so it cannot carry the session's authority past
    # the session: an insert into its source is refused instead.
    host = program("hanzo-cloud")
    ok("CREATE TABLE hanzo.definer_source (organization String, tokens UInt32) ENGINE = MergeTree ORDER BY organization", host)
    ok("CREATE TABLE hanzo.definer_target (organization String, tokens UInt64) ENGINE = SummingMergeTree ORDER BY organization", host)
    ok("CREATE MATERIALIZED VIEW hanzo.definer_mv TO hanzo.definer_target AS SELECT organization, toUInt64(tokens) AS tokens FROM hanzo.definer_source", host)
    assert "admin/hanzo-cloud:definer" in ok("SHOW CREATE TABLE hanzo.definer_mv", host)
    assert "ACCESS_DENIED" in denied("INSERT INTO hanzo.definer_source VALUES ('acme', 1)", host)
    ok("DROP TABLE hanzo.definer_mv", host)
    ok("INSERT INTO hanzo.definer_source VALUES ('acme', 1)", host)
    ok("DROP TABLE hanzo.definer_source", host)
    ok("DROP TABLE hanzo.definer_target", host)


def test_schema_role_is_scoped_to_its_database(started_cluster):
    schema = program("hanzo-o11y")
    # Its own plane, every row: a migration that backfills must see them all.
    assert ok("SELECT count() FROM event.fact", schema) == f"{FACTS}\n"
    # Nothing outside it, and no role it does not own.
    assert "ACCESS_DENIED" in denied("CREATE TABLE default.escape (x UInt8) ENGINE = Memory", schema)
    assert "ACCESS_DENIED" in denied("CREATE TABLE hanzo.escape (x UInt8) ENGINE = Memory", schema)
    assert "ACCESS_DENIED" in denied("CREATE ROLE intruder", schema)
    assert "ACCESS_DENIED" in denied("GRANT superadmin TO reader", schema)
    assert "ACCESS_DENIED" in denied("GRANT SELECT ON system.* TO reader", schema)


def test_no_object_reaches_past_its_database(started_cluster):
    # A dictionary's source logs in as a user of its own choosing (by default the loopback `default`),
    # and Distributed, Merge, Buffer, URL and the database engines read another database or server.
    # Neither the schema's owner nor the host may create one: no CREATE DICTIONARY, and table engines
    # need a grant (datastore-iam.xml), of which they hold only the MergeTree family and Null.
    host = program("hanzo-cloud")
    ok("CREATE TABLE IF NOT EXISTS hanzo.cloud_usage (organization String, tokens UInt32) ENGINE = MergeTree ORDER BY organization", host)
    for bearer, own, other, table in [
        (program("hanzo-o11y"), "event", "hanzo", "cloud_usage"),
        (host, "hanzo", "event", "fact"),
    ]:
        for sql in [
            f"CREATE DICTIONARY {own}.reach (k String, v String) PRIMARY KEY k SOURCE(DATASTORE(DB '{other}' TABLE '{table}')) LAYOUT(COMPLEX_KEY_HASHED()) LIFETIME(0)",
            f"CREATE DICTIONARY {own}.reach (k String, v String) PRIMARY KEY k SOURCE(HTTP(URL 'http://localhost:8123/' FORMAT 'TabSeparated')) LAYOUT(COMPLEX_KEY_HASHED()) LIFETIME(0)",
            f"CREATE TABLE {own}.reach (k String) ENGINE = Distributed('default', '{other}', '{table}')",
            f"CREATE TABLE {own}.reach (k String) ENGINE = Merge('{other}', '.*')",
            f"CREATE TABLE {own}.reach (k String) ENGINE = Buffer('{other}', '{table}', 1, 10, 100, 10000, 1000000, 10000000, 100000000)",
            f"CREATE TABLE {own}.reach (k String) ENGINE = URL('http://localhost:8123/', 'TabSeparated')",
            f"CREATE TABLE {own}.reach (k String) ENGINE = PostgreSQL('localhost:9005', '{other}', '{table}', 'default', '')",
            f"CREATE TABLE {own}.reach (k String) ENGINE = Memory",
            f"CREATE TABLE {own}.reach AS remote('localhost', '{other}', '{table}')",
            f"CREATE DATABASE IF NOT EXISTS {own} ENGINE = MySQL('localhost:9004', '{other}', 'default', '')",
        ]:
            assert "ACCESS_DENIED" in denied(sql, bearer), sql
        assert ok(f"SELECT count() FROM system.tables WHERE database = '{own}' AND name = 'reach'", bearer) == "0\n"
    # What the migrations and the runners create is still theirs.
    schema = program("hanzo-o11y")
    for engine in ["MergeTree", "ReplacingMergeTree", "AggregatingMergeTree", "SummingMergeTree"]:
        ok(f"CREATE TABLE event.engine_{engine} (org LowCardinality(String), v UInt64) ENGINE = {engine} ORDER BY org", schema)
        ok(f"DROP TABLE event.engine_{engine}", schema)
    ok("CREATE TABLE event.engine_null (held UInt8) ENGINE = Null", schema)
    ok("DROP TABLE event.engine_null", schema)


def test_schema_keeps_the_migration_ledger(started_cluster):
    # What cloud's datastore.Migrate runs, as the o11y application, in the schema's own database: the
    # ledger (org-first, like every table of the plane), then one lock generation held, renewed and
    # released. It needs no right outside `event`.
    schema = program("hanzo-o11y")
    ok("CREATE DATABASE IF NOT EXISTS event", schema)
    ok(
        "CREATE TABLE IF NOT EXISTS event.migration (org LowCardinality(String) DEFAULT '', set LowCardinality(String), "
        "version UInt32, name String, sum String, by String, at DateTime64(3) DEFAULT now64(3)) "
        "ENGINE = ReplacingMergeTree(at) ORDER BY (org, set, version)",
        schema,
    )
    lock = "SELECT name, comment FROM system.tables WHERE database = 'event' AND startsWith(name, 'migration_lock_o11y_') FORMAT TSV"
    assert ok(lock, schema) == ""
    ok("CREATE TABLE event.migration_lock_o11y_1 (held UInt8) ENGINE = Null COMMENT 'o11y-test until 0'", schema)
    assert ok(lock, schema) == "migration_lock_o11y_1\to11y-test until 0\n"
    ok("ALTER TABLE event.migration_lock_o11y_1 MODIFY COMMENT 'o11y-test until 1'", schema)
    ok("SELECT toUnixTimestamp(now()) AS now", schema)
    ok("INSERT INTO event.migration (set, version, name, sum, by) VALUES ('o11y', 5, '0005_iam_roles.sql', 'digest', 'o11y-test')", schema)
    assert ok("SELECT version, sum FROM event.migration FINAL WHERE set = 'o11y' FORMAT TSV", schema) == "5\tdigest\n"
    ok("DROP TABLE IF EXISTS event.migration_lock_o11y_1", schema)
    # A tenant sees none of the ledger (its org is ''), and the host does not touch it.
    assert ok("SELECT count() FROM event.migration", person("acme/alice", "acme")) == "0\n"
    assert "ACCESS_DENIED" in denied("SELECT count() FROM event.migration", program("hanzo-cloud"))
    # Nothing of the host's database is the schema's.
    assert "ACCESS_DENIED" in denied("CREATE TABLE IF NOT EXISTS hanzo.migration (x UInt8) ENGINE = Memory", schema)


def test_grants_reach_only_the_planes_roles(started_cluster):
    # The schema's grant option (and the host's) may name reader, writer and sink, and nobody else: not a
    # user of another directory, not another role. SuperAdmin's may name anyone.
    superadmin = person("admin/z", "admin", kid="cert-admin")
    ok("CREATE ROLE IF NOT EXISTS outsider", superadmin)
    ok("CREATE USER IF NOT EXISTS legacy HOST NONE", superadmin)
    schema = program("hanzo-o11y")
    for grantee in ["legacy", "outsider", "tenant", "legacy, reader"]:
        assert "is not allowed as grantee" in denied(f"GRANT SELECT ON event.* TO {grantee}", schema), grantee
    assert "is not allowed as grantee" in denied("GRANT SELECT ON hanzo.* TO legacy", program("hanzo-cloud"))
    ok("GRANT SELECT ON event.fact TO reader", schema)
    assert ok("SHOW GRANTS FOR legacy FORMAT TSV", superadmin) == ""
    assert ok("SHOW GRANTS FOR outsider FORMAT TSV", superadmin) == ""
    ok("GRANT SELECT ON event.fact TO outsider", superadmin)
    ok("DROP USER legacy", superadmin)
    ok("DROP ROLE outsider", superadmin)


def test_migration_in_one_session_grants_to_the_roles_it_creates(started_cluster):
    # Cloud's runner applies a migration over one pooled connection, logged in before the migration
    # creates its roles. The roles it creates are grantees of that login from the moment they exist.
    superadmin = person("admin/z", "admin", kid="cert-admin")
    ok("DROP ROLE IF EXISTS reader, writer, sink", superadmin)
    native(";\n".join(MIGRATION), program("hanzo-o11y"))
    assert ok("SHOW GRANTS FOR reader FORMAT TSV", superadmin) == "GRANT SELECT ON event.* TO reader\n"
    assert ok("SHOW GRANTS FOR writer FORMAT TSV", superadmin) == (
        "GRANT INSERT ON event.* TO writer\nREVOKE INSERT ON event.fact FROM writer\n"
    )
    assert ok("SHOW GRANTS FOR sink FORMAT TSV", superadmin) == "GRANT INSERT ON event.fact TO sink\n"
    assert ok("SELECT count() FROM event.fact", person("acme/alice", "acme")) == "2\n"


def test_query_cache_keeps_sessions_apart(started_cluster):
    # One subject, two tokens: SuperAdmin under the admin key, an ordinary admin-org person under
    # another. A result cached by the first must not answer the second.
    query = "SELECT count() FROM event.fact"
    cached = "?use_query_cache=1&query_cache_share_between_users=1"
    assert ok(query, person("admin/z", "admin", kid="cert-admin"), cached) == f"{FACTS}\n"
    assert ok(query, person("admin/z", "admin", kid="cert-hanzo"), cached) == "0\n"
    assert ok(query, person("admin/y", "admin", kid="cert-hanzo"), cached) == "0\n"
    # And an org's shared entry answers only that org.
    assert ok(query, person("acme/alice", "acme"), cached) == "2\n"
    assert ok(query, person("zoo/bob", "zoo"), cached) == "1\n"


def test_rejected_tokens(started_cluster):
    for bad in [
        person("acme/alice", "acme", aud="console"),
        person("acme/alice", "acme", exp_in="-3600"),
        person("acme/alice", "acme", kid="cert-unknown"),
        person("acme/alice", "acme", iss="http://evil.test"),
        person("acme/alice", "acme")[:-4] + "AAAA",
        "not-a-token",
    ]:
        code, body = http("SELECT 1", bad)
        assert code == "516" or "AUTHENTICATION_FAILED" in body, body


def test_bearer_does_not_mix_with_a_password(started_cluster):
    code, body = http("SELECT currentUser()", person("acme/alice", "acme"), "?user=default")
    assert code != "200" and "not allowed to use" in body, body


def test_native_protocol_jwt(started_cluster):
    bob = person("zoo/bob", "zoo")
    assert native("SELECT currentUser(), currentOrg() FORMAT TSV", bob) == "zoo/bob\tzoo\n"
    assert native("SELECT count() FROM event.fact", bob) == "1\n"


def test_session_roles_cannot_be_shed(started_cluster):
    # The token's roles are the session's, not grants of the user: SET ROLE cannot drop `tenant`
    # and with it the confinement.
    bob = person("zoo/bob", "zoo")
    assert native("SET ROLE NONE; SELECT count() FROM event.fact", bob) == "1\n"
    assert native("SET ROLE DEFAULT; SELECT count() FROM event.fact", bob) == "1\n"