mirror of
https://github.com/simonw/datasette.git
synced 2026-07-08 16:44:34 +02:00
aaaffe45b8Invalid dstok_ tokens - bad signature, malformed payload, expired, or presented while allow_signed_tokens is off - previously degraded the request to anonymous, so clients saw a 403 permission error or worse, a 200 with anonymous-visible data. Token handlers can now raise TokenInvalid for tokens they recognize but reject; Datasette responds with 401, the canonical JSON error body and a WWW-Authenticate: Bearer error="invalid_token" header, even when a valid cookie is also present. Bearer tokens no registered handler recognizes are still ignored, so authentication plugins with their own token formats keep working. TokenInvalid is exported from the datasette package for use by plugin token handlers. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GrHZSypDfMnym1tM5XJAFZ
344 lines
11 KiB
Python
344 lines
11 KiB
Python
"""
|
|
Tests for the register_token_handler plugin hook.
|
|
"""
|
|
|
|
from datasette.app import Datasette
|
|
from datasette.hookspecs import hookimpl
|
|
from datasette.plugins import pm
|
|
from datasette.tokens import (
|
|
TokenHandler,
|
|
TokenInvalid,
|
|
TokenRestrictions,
|
|
SignedTokenHandler,
|
|
)
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def datasette():
|
|
return Datasette()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_signed_handler_registered(datasette):
|
|
"""The default SignedTokenHandler should be registered automatically."""
|
|
handlers = datasette._token_handlers()
|
|
assert len(handlers) >= 1
|
|
assert any(isinstance(h, SignedTokenHandler) for h in handlers)
|
|
assert any(h.name == "signed" for h in handlers)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_default(datasette):
|
|
"""create_token() with handler='signed' should create a signed token."""
|
|
token = await datasette.create_token("test_actor", handler="signed")
|
|
assert token.startswith("dstok_")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_with_restrictions(datasette):
|
|
"""create_token() should handle restriction parameters."""
|
|
token = await datasette.create_token(
|
|
"test_actor",
|
|
handler="signed",
|
|
expires_after=3600,
|
|
restrictions=TokenRestrictions().allow_all("view-instance"),
|
|
)
|
|
assert token.startswith("dstok_")
|
|
# Verify the token contains the expected data
|
|
decoded = datasette.unsign(token[len("dstok_") :], namespace="token")
|
|
assert decoded["a"] == "test_actor"
|
|
assert decoded["d"] == 3600
|
|
assert "_r" in decoded
|
|
assert "a" in decoded["_r"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_default(datasette):
|
|
"""verify_token() should verify signed tokens."""
|
|
token = await datasette.create_token("test_actor", handler="signed")
|
|
actor = await datasette.verify_token(token)
|
|
assert actor is not None
|
|
assert actor["id"] == "test_actor"
|
|
assert actor["token"] == "dstok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_unknown_returns_none(datasette):
|
|
"""verify_token() should return None for unrecognized tokens."""
|
|
result = await datasette.verify_token("unknown_token_format_xyz")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_bad_signature_raises(datasette):
|
|
"""verify_token() should raise TokenInvalid for tokens with bad signatures."""
|
|
with pytest.raises(TokenInvalid):
|
|
await datasette.verify_token("dstok_tampered_data_here")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_with_named_handler(datasette):
|
|
"""create_token(handler='signed') should select the signed handler."""
|
|
token = await datasette.create_token("test_actor", handler="signed")
|
|
assert token.startswith("dstok_")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_unknown_handler_raises(datasette):
|
|
"""create_token(handler='nonexistent') should raise ValueError."""
|
|
with pytest.raises(ValueError, match="Token handler 'nonexistent' not found"):
|
|
await datasette.create_token("test_actor", handler="nonexistent")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_token_handler(datasette):
|
|
"""A custom token handler should be usable for both create and verify."""
|
|
|
|
class CustomHandler(TokenHandler):
|
|
name = "custom"
|
|
|
|
async def create_token(self, datasette, actor_id, **kwargs):
|
|
return f"custom_{actor_id}"
|
|
|
|
async def verify_token(self, datasette, token):
|
|
if token.startswith("custom_"):
|
|
return {"id": token[len("custom_") :], "token": "custom"}
|
|
return None
|
|
|
|
class Plugin:
|
|
__name__ = "CustomTokenPlugin"
|
|
|
|
@staticmethod
|
|
@hookimpl
|
|
def register_token_handler(datasette):
|
|
return CustomHandler()
|
|
|
|
pm.register(Plugin(), name="test_custom_handler")
|
|
try:
|
|
handlers = datasette._token_handlers()
|
|
assert any(h.name == "custom" for h in handlers)
|
|
|
|
# Create with custom handler
|
|
token = await datasette.create_token("alice", handler="custom")
|
|
assert token == "custom_alice"
|
|
|
|
# Verify custom token
|
|
actor = await datasette.verify_token("custom_alice")
|
|
assert actor is not None
|
|
assert actor["id"] == "alice"
|
|
assert actor["token"] == "custom"
|
|
|
|
# Signed tokens should still work
|
|
signed_token = await datasette.create_token("bob", handler="signed")
|
|
assert signed_token.startswith("dstok_")
|
|
actor = await datasette.verify_token(signed_token)
|
|
assert actor["id"] == "bob"
|
|
finally:
|
|
pm.unregister(name="test_custom_handler")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_token_tries_all_handlers(datasette):
|
|
"""verify_token() should try each handler until one matches."""
|
|
|
|
class HandlerA(TokenHandler):
|
|
name = "handler_a"
|
|
|
|
async def create_token(self, datasette, actor_id, **kwargs):
|
|
return f"a_{actor_id}"
|
|
|
|
async def verify_token(self, datasette, token):
|
|
if token.startswith("a_"):
|
|
return {"id": token[2:], "token": "handler_a"}
|
|
return None
|
|
|
|
class HandlerB(TokenHandler):
|
|
name = "handler_b"
|
|
|
|
async def create_token(self, datasette, actor_id, **kwargs):
|
|
return f"b_{actor_id}"
|
|
|
|
async def verify_token(self, datasette, token):
|
|
if token.startswith("b_"):
|
|
return {"id": token[2:], "token": "handler_b"}
|
|
return None
|
|
|
|
class PluginA:
|
|
__name__ = "PluginA"
|
|
|
|
@staticmethod
|
|
@hookimpl
|
|
def register_token_handler(datasette):
|
|
return HandlerA()
|
|
|
|
class PluginB:
|
|
__name__ = "PluginB"
|
|
|
|
@staticmethod
|
|
@hookimpl
|
|
def register_token_handler(datasette):
|
|
return HandlerB()
|
|
|
|
pm.register(PluginA(), name="test_handler_a")
|
|
pm.register(PluginB(), name="test_handler_b")
|
|
try:
|
|
# Both handler tokens should verify
|
|
actor_a = await datasette.verify_token("a_alice")
|
|
assert actor_a is not None
|
|
assert actor_a["id"] == "alice"
|
|
assert actor_a["token"] == "handler_a"
|
|
|
|
actor_b = await datasette.verify_token("b_bob")
|
|
assert actor_b is not None
|
|
assert actor_b["id"] == "bob"
|
|
assert actor_b["token"] == "handler_b"
|
|
|
|
# Unknown token should return None
|
|
assert await datasette.verify_token("c_charlie") is None
|
|
finally:
|
|
pm.unregister(name="test_handler_a")
|
|
pm.unregister(name="test_handler_b")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_handler_via_http(datasette):
|
|
"""Default signed tokens should work through HTTP auth."""
|
|
token = await datasette.create_token("http_user", handler="signed")
|
|
response = await datasette.client.get(
|
|
"/-/actor.json",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
actor = response.json()["actor"]
|
|
assert actor["id"] == "http_user"
|
|
assert actor["token"] == "dstok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_handler_via_http(datasette):
|
|
"""Custom handler tokens should work through HTTP auth."""
|
|
|
|
class CustomHandler(TokenHandler):
|
|
name = "custom_http"
|
|
|
|
async def create_token(self, datasette, actor_id, **kwargs):
|
|
return f"chttp_{actor_id}"
|
|
|
|
async def verify_token(self, datasette, token):
|
|
if token.startswith("chttp_"):
|
|
return {"id": token[len("chttp_") :], "token": "custom_http"}
|
|
return None
|
|
|
|
class Plugin:
|
|
__name__ = "CustomHTTPPlugin"
|
|
|
|
@staticmethod
|
|
@hookimpl
|
|
def register_token_handler(datasette):
|
|
return CustomHandler()
|
|
|
|
pm.register(Plugin(), name="test_custom_http")
|
|
try:
|
|
token = await datasette.create_token("web_user", handler="custom_http")
|
|
response = await datasette.client.get(
|
|
"/-/actor.json",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
actor = response.json()["actor"]
|
|
assert actor["id"] == "web_user"
|
|
assert actor["token"] == "custom_http"
|
|
finally:
|
|
pm.unregister(name="test_custom_http")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_handler_base_class_raises():
|
|
"""TokenHandler base class methods should raise NotImplementedError."""
|
|
handler = TokenHandler()
|
|
ds = Datasette()
|
|
with pytest.raises(NotImplementedError):
|
|
await handler.create_token(ds, "test")
|
|
with pytest.raises(NotImplementedError):
|
|
await handler.verify_token(ds, "test")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restrictions_round_trip(datasette):
|
|
"""Tokens with database/resource restrictions should round-trip correctly."""
|
|
restrictions = (
|
|
TokenRestrictions()
|
|
.allow_all("view-instance")
|
|
.allow_database("docs", "view-query")
|
|
.allow_resource("docs", "attachments", "insert-row")
|
|
)
|
|
token = await datasette.create_token(
|
|
"test_actor", handler="signed", restrictions=restrictions
|
|
)
|
|
actor = await datasette.verify_token(token)
|
|
assert actor is not None
|
|
assert actor["id"] == "test_actor"
|
|
assert actor["_r"]["a"] == ["view-instance"]
|
|
assert actor["_r"]["d"] == {"docs": ["view-query"]}
|
|
assert actor["_r"]["r"] == {"docs": {"attachments": ["insert-row"]}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expires_after_round_trip(datasette):
|
|
"""Tokens with expires_after should include token_expires in the actor."""
|
|
token = await datasette.create_token(
|
|
"test_actor", handler="signed", expires_after=3600
|
|
)
|
|
actor = await datasette.verify_token(token)
|
|
assert actor is not None
|
|
assert actor["id"] == "test_actor"
|
|
assert "token_expires" in actor
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"build_restrictions,expected",
|
|
[
|
|
(lambda r: r, None),
|
|
(lambda r: r.allow_all("view-instance"), {"a": ["vi"]}),
|
|
(
|
|
lambda r: r.allow_database("docs", "view-query"),
|
|
{"d": {"docs": ["vq"]}},
|
|
),
|
|
(
|
|
lambda r: r.allow_resource("docs", "attachments", "insert-row"),
|
|
{"r": {"docs": {"attachments": ["ir"]}}},
|
|
),
|
|
(
|
|
lambda r: r.allow_all("view-instance")
|
|
.allow_database("docs", "view-query")
|
|
.allow_resource("docs", "attachments", "insert-row"),
|
|
{
|
|
"a": ["vi"],
|
|
"d": {"docs": ["vq"]},
|
|
"r": {"docs": {"attachments": ["ir"]}},
|
|
},
|
|
),
|
|
(
|
|
lambda r: r.allow_all("not-a-real-action"),
|
|
{"a": ["not-a-real-action"]},
|
|
),
|
|
],
|
|
ids=["empty", "all", "database", "resource", "combined", "unknown_action"],
|
|
)
|
|
async def test_token_restrictions_abbreviated(datasette, build_restrictions, expected):
|
|
await datasette.invoke_startup()
|
|
restrictions = build_restrictions(TokenRestrictions())
|
|
assert restrictions.abbreviated(datasette) == expected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signed_tokens_disabled():
|
|
"""create_token and verify_token should fail/skip when signed tokens are disabled."""
|
|
ds = Datasette(settings={"allow_signed_tokens": False})
|
|
with pytest.raises(ValueError, match="Signed tokens are not enabled"):
|
|
await ds.create_token("test_actor", handler="signed")
|
|
# verify_token should raise TokenInvalid for a dstok_ token
|
|
with pytest.raises(TokenInvalid, match="not enabled"):
|
|
await ds.verify_token("dstok_anything")
|