menzeljandClaude Opus 5
e650aa6833
A session token is the wrong credential for automation. It expires in an hour, it is minted by typing a password, and revoking it signs every one of that person's devices out. So automation gets its own credential, revocable on its own, and showing up in the audit log as itself. Three decisions worth recording, because each one is a place this could have been built wrong. **Only a hash is stored.** This is the opposite call from registry passwords one release ago, and for a concrete reason: a registry password has to be handed back to the registry, so it must be recoverable and is encrypted. A token is only ever compared against, so it does not need to be — and not keeping it is the difference between leaking the database and leaking everything the database protects. It is shown once and cannot be recovered; a readable prefix is kept so rows are still identifiable in the UI and the audit log. The hash is SHA-256, deliberately not bcrypt: bcrypt is slow to make guessing low-entropy human passwords expensive, and a token is 256 bits of secrets output, so the cost would buy nothing and would land on every single API request. **The scope is not folded into the User object.** get_current_user returns a session-attached row; downgrading its role in place to represent a read-only token would be written back to the database the next time anything committed that user — logout-everywhere does exactly that. So the token row is stashed on request.state and require_admin consults it, leaving the User untouched. The same lookup caps a token at its owner's authority rather than trusting the scope alone, so a demoted admin's token drops to read-only with them and a disabled account's tokens stop working. **A token cannot make itself permanent.** Creating tokens and creating users now require a signed-in session, via a require_session dependency that rejects token-authenticated requests. Without it, a leaked CI credential could mint a second one and survive its own revocation — the failure mode where revoking the leak does nothing. This is the one behaviour change for existing installs: scripted user creation now needs a login. The WebSocket routes still take JWTs only. They carry logs, the terminal and the deploy console, which a CI job has no use for, and leaving them alone keeps the token surface to the REST API. 19 tests, covering what is stored, that a read token really is read-only while its owner is an admin, that demoting and disabling the owner both take effect, expiry, tampering, the throttle on last-used writes, and that a token can neither mint another nor create a user. Verified end to end against a running app: two tokens, both scopes, revocation, and no plaintext anywhere in the database or the list response. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""API tokens: long-lived credentials for scripts and CI.
|
|
|
|
A token is a credential that outlives a session and is handed to a machine, so
|
|
the things worth pinning down are the ones that go wrong quietly:
|
|
|
|
* only a hash is stored, and the token is returned exactly once,
|
|
* a read-scoped token really is read-only, even when its owner is an admin,
|
|
* a token never outranks its owner — demote or disable the account and the
|
|
token follows,
|
|
* and a token cannot mint another one, so a leak cannot be made permanent.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlmodel import Session, delete, select
|
|
|
|
|
|
@pytest.fixture
|
|
def svc(db):
|
|
from services import api_token_service
|
|
|
|
return api_token_service
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_tokens(db):
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
with Session(engine) as session:
|
|
session.exec(delete(ApiToken))
|
|
session.commit()
|
|
yield
|
|
with Session(engine) as session:
|
|
session.exec(delete(ApiToken))
|
|
session.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
def owner(db):
|
|
"""A throwaway admin, so tests can disable or demote it freely."""
|
|
from database import engine
|
|
from models.user import User
|
|
import auth as auth_mod
|
|
|
|
with Session(engine) as session:
|
|
existing = session.exec(select(User).where(User.username == "token-owner")).first()
|
|
if existing:
|
|
session.delete(existing)
|
|
session.commit()
|
|
user = User(
|
|
username="token-owner",
|
|
hashed_password=auth_mod.hash_password("pw"),
|
|
role="admin",
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
session.refresh(user)
|
|
user_id = user.id
|
|
yield user_id
|
|
with Session(engine) as session:
|
|
row = session.get(User, user_id)
|
|
if row:
|
|
session.delete(row)
|
|
session.commit()
|
|
|
|
|
|
def _mint(svc, owner_id: int, scope: str = "admin", expires_in_days=None) -> str:
|
|
from database import engine
|
|
from models.user import User
|
|
|
|
with Session(engine) as session:
|
|
user = session.get(User, owner_id)
|
|
_row, token = svc.mint(
|
|
session, name=f"test-{scope}", user=user, scope=scope,
|
|
expires_in_days=expires_in_days,
|
|
)
|
|
return token
|
|
|
|
|
|
def _as(client, token: str):
|
|
"""A tiny client that authenticates with a raw bearer token."""
|
|
|
|
class _Client:
|
|
def get(self, url, **kw):
|
|
return client.get(url, headers={"Authorization": f"Bearer {token}"}, **kw)
|
|
|
|
def post(self, url, **kw):
|
|
return client.post(url, headers={"Authorization": f"Bearer {token}"}, **kw)
|
|
|
|
def delete(self, url, **kw):
|
|
return client.delete(url, headers={"Authorization": f"Bearer {token}"}, **kw)
|
|
|
|
return _Client()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# What is stored
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_the_token_is_stored_only_as_a_hash(svc, owner):
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
token = _mint(svc, owner)
|
|
with Session(engine) as session:
|
|
row = session.exec(select(ApiToken)).one()
|
|
assert token not in row.token_hash
|
|
assert row.token_hash != token
|
|
# The readable front is kept so the UI can name the row, and it is far too
|
|
# short to authenticate with.
|
|
assert token.startswith(row.prefix)
|
|
assert len(row.prefix) < len(token) / 2
|
|
|
|
|
|
def test_tokens_are_unique_and_prefixed(svc, owner):
|
|
first, second = _mint(svc, owner), _mint(svc, owner)
|
|
assert first != second
|
|
assert first.startswith("sp_") and second.startswith("sp_")
|
|
|
|
|
|
def test_a_tampered_token_does_not_resolve(svc, owner):
|
|
from database import engine
|
|
|
|
token = _mint(svc, owner)
|
|
with Session(engine) as session:
|
|
assert svc.resolve(session, token) is not None
|
|
assert svc.resolve(session, token[:-1] + "x") is None
|
|
assert svc.resolve(session, "sp_totallymadeupvalue") is None
|
|
# A JWT must not be mistaken for one.
|
|
assert svc.resolve(session, "eyJhbGciOiJIUzI1NiJ9.e30.x") is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Using one
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_a_token_authenticates_like_a_session(client, svc, owner):
|
|
response = _as(client, _mint(svc, owner)).get("/api/auth/me")
|
|
assert response.status_code == 200
|
|
assert response.json()["username"] == "token-owner"
|
|
|
|
|
|
def test_a_read_token_cannot_change_anything(client, svc, owner):
|
|
"""The owner is an admin; the token is not. That is the point of scopes."""
|
|
api = _as(client, _mint(svc, owner, scope="read"))
|
|
assert api.get("/api/stacks").status_code == 200
|
|
denied = api.post("/api/registries", json={"host": "ghcr.io", "username": "a", "password": "b"})
|
|
assert denied.status_code == 403
|
|
assert "read-only" in denied.json()["detail"]
|
|
|
|
|
|
def test_an_admin_token_may_act(client, svc, owner):
|
|
api = _as(client, _mint(svc, owner, scope="admin"))
|
|
created = api.post(
|
|
"/api/registries",
|
|
json={"host": "quay.io", "username": "a", "password": "b"},
|
|
)
|
|
assert created.status_code == 201, created.text
|
|
api.delete(f"/api/registries/{created.json()['id']}")
|
|
|
|
|
|
def test_using_a_token_records_when(client, svc, owner):
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
_as(client, _mint(svc, owner)).get("/api/auth/me")
|
|
with Session(engine) as session:
|
|
assert session.exec(select(ApiToken)).one().last_used_at is not None
|
|
|
|
|
|
def test_the_last_used_write_is_throttled(svc, owner, monkeypatch):
|
|
"""Otherwise every API call is also a database write."""
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
token = _mint(svc, owner)
|
|
with Session(engine) as session:
|
|
row = svc.resolve(session, token)[0]
|
|
svc.touch(session, row)
|
|
first = row.last_used_at
|
|
svc.touch(session, row)
|
|
assert row.last_used_at == first
|
|
|
|
# Far enough in the past and it is written again.
|
|
row.last_used_at = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
session.add(row)
|
|
session.commit()
|
|
svc.touch(session, row)
|
|
assert session.exec(select(ApiToken)).one().last_used_at != first
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# A token never outranks its owner
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_a_demoted_owner_drops_the_token_to_read_only(client, svc, owner):
|
|
from database import engine
|
|
from models.user import User
|
|
|
|
token = _mint(svc, owner, scope="admin")
|
|
with Session(engine) as session:
|
|
user = session.get(User, owner)
|
|
user.role = "user"
|
|
session.add(user)
|
|
session.commit()
|
|
|
|
api = _as(client, token)
|
|
assert api.get("/api/stacks").status_code == 200
|
|
assert api.post("/api/registries", json={"host": "x.io", "username": "a", "password": "b"}).status_code == 403
|
|
|
|
|
|
def test_a_disabled_owner_kills_the_token(client, svc, owner):
|
|
from database import engine
|
|
from models.user import User
|
|
|
|
token = _mint(svc, owner)
|
|
with Session(engine) as session:
|
|
user = session.get(User, owner)
|
|
user.is_active = False
|
|
session.add(user)
|
|
session.commit()
|
|
|
|
assert _as(client, token).get("/api/auth/me").status_code == 401
|
|
|
|
|
|
def test_an_expired_token_is_refused(client, svc, owner):
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
token = _mint(svc, owner, expires_in_days=1)
|
|
with Session(engine) as session:
|
|
row = session.exec(select(ApiToken)).one()
|
|
row.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
response = _as(client, token).get("/api/auth/me")
|
|
assert response.status_code == 401
|
|
assert "not valid" in response.json()["detail"]
|
|
|
|
|
|
def test_a_token_with_no_expiry_does_not_expire(svc, owner):
|
|
from database import engine
|
|
from models.api_token import ApiToken
|
|
|
|
_mint(svc, owner)
|
|
with Session(engine) as session:
|
|
assert svc.is_expired(session.exec(select(ApiToken)).one()) is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# A token cannot make itself permanent
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_a_token_cannot_mint_another_token(client, svc, owner):
|
|
api = _as(client, _mint(svc, owner, scope="admin"))
|
|
response = api.post("/api/auth/tokens", json={"name": "second", "scope": "admin"})
|
|
assert response.status_code == 403
|
|
assert "signed-in session" in response.json()["detail"]
|
|
|
|
|
|
def test_a_token_cannot_create_a_user(client, svc, owner):
|
|
api = _as(client, _mint(svc, owner, scope="admin"))
|
|
response = api.post(
|
|
"/api/auth/users", json={"username": "backdoor", "password": "x", "role": "admin"}
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_a_token_cannot_even_list_tokens(client, svc, owner):
|
|
assert _as(client, _mint(svc, owner, scope="admin")).get("/api/auth/tokens").status_code == 403
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Through the API
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_create_shows_the_token_once_and_never_again(as_admin):
|
|
created = as_admin.post("/api/auth/tokens", json={"name": "ci", "scope": "read"})
|
|
assert created.status_code == 201, created.text
|
|
token = created.json()["token"]
|
|
assert token.startswith("sp_")
|
|
|
|
listed = as_admin.get("/api/auth/tokens")
|
|
assert token not in listed.text
|
|
assert '"token"' not in listed.text
|
|
assert listed.json()[0]["prefix"] == token[:11]
|
|
|
|
|
|
def test_revoking_a_token_stops_it_working(client, as_admin):
|
|
created = as_admin.post("/api/auth/tokens", json={"name": "ci", "scope": "read"}).json()
|
|
api = _as(client, created["token"])
|
|
assert api.get("/api/auth/me").status_code == 200
|
|
|
|
assert as_admin.delete(f"/api/auth/tokens/{created['id']}").status_code == 200
|
|
assert api.get("/api/auth/me").status_code == 401
|
|
|
|
|
|
def test_a_bad_scope_or_empty_name_is_refused(as_admin):
|
|
assert as_admin.post("/api/auth/tokens", json={"name": "x", "scope": "root"}).status_code == 400
|
|
assert as_admin.post("/api/auth/tokens", json={"name": " ", "scope": "read"}).status_code == 400
|
|
assert (
|
|
as_admin.post(
|
|
"/api/auth/tokens", json={"name": "x", "scope": "read", "expires_in_days": 0}
|
|
).status_code
|
|
== 400
|
|
)
|
|
|
|
|
|
def test_the_read_only_role_cannot_manage_tokens(as_user):
|
|
assert as_user.get("/api/auth/tokens").status_code == 403
|
|
assert as_user.post("/api/auth/tokens", json={"name": "x"}).status_code == 403
|