mirror
/
datasette
kopia lustrzana https://github.com/simonw/datasette
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

max_signed_tokens_ttl setting, closes #1858 

Also redesigned token format to include creation time and optional duration.
pull/1912/head
Simon Willison 2022-10-26 14:13:31 -07:00
rodzic af5d5d0243
commit 382a871583
6 zmienionych plików z 99 dodań i 25 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

5
datasette/app.py
Wyświetl plik

@ -129,6 +129,11 @@ SETTINGS = (
True,
"Allow users to create and use signed API tokens",
),
Setting(
"max_signed_tokens_ttl",
0,
"Maximum allowed expiry time for signed API tokens",
),
Setting("suggest_facets", True, "Calculate and display suggested facets"),
Setting(
"default_cache_ttl",

33
datasette/default_permissions.py
Wyświetl plik

@ -56,6 +56,7 @@ def actor_from_request(datasette, request):
prefix = "dstok_"
if not datasette.setting("allow_signed_tokens"):
return None
max_signed_tokens_ttl = datasette.setting("max_signed_tokens_ttl")
authorization = request.headers.get("authorization")
if not authorization:
return None
@ -69,11 +70,31 @@ def actor_from_request(datasette, request):
decoded = datasette.unsign(token, namespace="token")
except itsdangerous.BadSignature:
return None
expires_at = decoded.get("e")
if expires_at is not None:
if expires_at < time.time():
if "t" not in decoded:
# Missing timestamp
return None
created = decoded["t"]
if not isinstance(created, int):
# Invalid timestamp
return None
duration = decoded.get("d")
if duration is not None and not isinstance(duration, int):
# Invalid duration
return None
if (duration is None and max_signed_tokens_ttl) or (
duration is not None
and max_signed_tokens_ttl
and duration > max_signed_tokens_ttl
):
duration = max_signed_tokens_ttl
if duration:
if time.time() - created > duration:
# Expired
return None
return {"id": decoded["a"], "token": "dstok"}
actor = {"id": decoded["a"], "token": "dstok"}
if duration:
actor["token_expires"] = created + duration
return actor
@hookimpl
@ -102,9 +123,9 @@ def register_commands(cli):
def create_token(id, secret, expires_after, debug):
"Create a signed API token for the specified actor ID"
ds = Datasette(secret=secret)
bits = {"a": id, "token": "dstok"}
bits = {"a": id, "token": "dstok", "t": int(time.time())}
if expires_after:
bits["e"] = int(time.time()) + expires_after
bits["d"] = expires_after
token = ds.sign(bits, namespace="token")
click.echo("dstok_{}".format(token))
if debug:

20
datasette/views/special.py
Wyświetl plik

@ -195,20 +195,24 @@ class CreateTokenView(BaseView):
async def post(self, request):
self.check_permission(request)
post = await request.post_vars()
expires = None
errors = []
duration = None
if post.get("expire_type"):
duration = post.get("expire_duration")
if not duration or not duration.isdigit() or not int(duration) > 0:
duration_string = post.get("expire_duration")
if (
not duration_string
or not duration_string.isdigit()
or not int(duration_string) > 0
):
errors.append("Invalid expire duration")
else:
unit = post["expire_type"]
if unit == "minutes":
expires = int(duration) * 60
duration = int(duration_string) * 60
elif unit == "hours":
expires = int(duration) * 60 * 60
duration = int(duration_string) * 60 * 60
elif unit == "days":
expires = int(duration) * 60 * 60 * 24
duration = int(duration_string) * 60 * 60 * 24
else:
errors.append("Invalid expire duration unit")
token_bits = None
@ -216,8 +220,10 @@ class CreateTokenView(BaseView):
if not errors:
token_bits = {
"a": request.actor["id"],
"e": (int(time.time()) + expires) if expires else None,
"t": int(time.time()),
}
if duration:
token_bits["d"] = duration
token = "dstok_{}".format(self.ds.sign(token_bits, "token"))
return await self.render(
["create_token.html"],

15
docs/settings.rst
Wyświetl plik

@ -182,6 +182,21 @@ This is turned on by default. Use the following to turn it off::
Turning this setting off will disable the ``/-/create-token`` page, :ref:`described here <CreateTokenView>`. It will also cause any incoming ``Authorization: Bearer dstok_...`` API tokens to be ignored.
.. _setting_max_signed_tokens_ttl:
max_signed_tokens_ttl
~~~~~~~~~~~~~~~~~~~~~
Maximum allowed expiry time for signed API tokens created by users.
Defaults to ``0`` which means no limit - tokens can be created that will never expire.
Set this to a value in seconds to limit the maximum expiry time. For example, to set that limit to 24 hours you would use::
datasette mydatabase.db --setting max_signed_tokens_ttl 86400
This setting is enforced when incoming tokens are processed.
.. _setting_default_cache_ttl:
default_cache_ttl

1
tests/test_api.py
Wyświetl plik

@ -807,6 +807,7 @@ def test_settings_json(app_client):
"sql_time_limit_ms": 200,
"allow_download": True,
"allow_signed_tokens": True,
"max_signed_tokens_ttl": 0,
"allow_facet": True,
"suggest_facets": True,
"default_cache_ttl": 5,

50
tests/test_auth.py
Wyświetl plik

@ -173,13 +173,19 @@ def test_auth_create_token(app_client, post_data, errors, expected_duration):
# Extract token from page
token = response2.text.split('value="dstok_')[1].split('"')[0]
details = app_client.ds.unsign(token, "token")
assert details.keys() == {"a", "e"}
assert details.keys() == {"a", "t", "d"} or details.keys() == {"a", "t"}
assert details["a"] == "test"
if expected_duration is None:
assert details["e"] is None
assert "d" not in details
else:
about_right = int(time.time()) + expected_duration
assert about_right - 2 < details["e"] < about_right + 2
assert details["d"] == expected_duration
# And test that token
response3 = app_client.get(
"/-/actor.json",
headers={"Authorization": "Bearer {}".format("dstok_{}".format(token))},
)
assert response3.status == 200
assert response3.json["actor"]["id"] == "test"
def test_auth_create_token_not_allowed_for_tokens(app_client):
@ -206,6 +212,7 @@ def test_auth_create_token_not_allowed_if_allow_signed_tokens_off(app_client):
(
("allow_signed_tokens_off", False),
("no_token", False),
("no_timestamp", False),
("invalid_token", False),
("expired_token", False),
("valid_unlimited_token", True),
@ -214,12 +221,15 @@ def test_auth_create_token_not_allowed_if_allow_signed_tokens_off(app_client):
)
def test_auth_with_dstok_token(app_client, scenario, should_work):
token = None
_time = int(time.time())
if scenario in ("valid_unlimited_token", "allow_signed_tokens_off"):
token = app_client.ds.sign({"a": "test"}, "token")
token = app_client.ds.sign({"a": "test", "t": _time}, "token")
elif scenario == "valid_expiring_token":
token = app_client.ds.sign({"a": "test", "e": int(time.time()) + 1000}, "token")
token = app_client.ds.sign({"a": "test", "t": _time - 50, "d": 1000}, "token")
elif scenario == "expired_token":
token = app_client.ds.sign({"a": "test", "e": int(time.time()) - 1000}, "token")
token = app_client.ds.sign({"a": "test", "t": _time - 2000, "d": 1000}, "token")
elif scenario == "no_timestamp":
token = app_client.ds.sign({"a": "test"}, "token")
elif scenario == "invalid_token":
token = "invalid"
if token:
@ -232,7 +242,16 @@ def test_auth_with_dstok_token(app_client, scenario, should_work):
response = app_client.get("/-/actor.json", headers=headers)
try:
if should_work:
assert response.json == {"actor": {"id": "test", "token": "dstok"}}
assert response.json.keys() == {"actor"}
actor = response.json["actor"]
expected_keys = {"id", "token"}
if scenario != "valid_unlimited_token":
expected_keys.add("token_expires")
assert actor.keys() == expected_keys
assert actor["id"] == "test"
assert actor["token"] == "dstok"
if scenario != "valid_unlimited_token":
assert isinstance(actor["token_expires"], int)
else:
assert response.json == {"actor": None}
finally:
@ -251,15 +270,22 @@ def test_cli_create_token(app_client, expires):
token = result.output.strip()
assert token.startswith("dstok_")
details = app_client.ds.unsign(token[len("dstok_") :], "token")
expected_keys = {"a", "token"}
expected_keys = {"a", "token", "t"}
if expires:
expected_keys.add("e")
expected_keys.add("d")
assert details.keys() == expected_keys
assert details["a"] == "test"
response = app_client.get(
"/-/actor.json", headers={"Authorization": "Bearer {}".format(token)}
)
if expires is None or expires > 0:
assert response.json == {"actor": {"id": "test", "token": "dstok"}}
expected_actor = {
"id": "test",
"token": "dstok",
}
if expires and expires > 0:
expected_actor["token_expires"] = details["t"] + expires
assert response.json == {"actor": expected_actor}
else:
assert response.json == {"actor": None}
expected_actor = None
assert response.json == {"actor": expected_actor}