mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
d56b48c499
Minimal OAuth 2.0 authorization-code flow with PKCE (S256): admins register an app with one exact callback URL (Settings > API Keys > Connected Apps); /connect/authorize asks for consent once and returns a single-use, 60 s code bound to app, callback and challenge; POST /api/v1/connect/token swaps it, with the client secret, for the user's identity and permissions. Codes and secrets stored hashed, exchanges rate-limited per client and IP, no redirect before the callback is validated, API keys cannot authorize, refused while auth is disabled. i18n for all 15 locales. ----- fix(db): upgrading from 0.2.4.0 or older no longer crashes at startup The #2974 failure-reason conversion ran before the #1378 migration that adds print_log_entries.failure_reason, so older databases stopped with "no such column: failure_reason". It now skips a table without the column, only runs where a legacy label exists, and on SQLite rebuilds archive_fts first, since archives created before that index existed trip "database disk image is malformed" when updated.
364 lines
16 KiB
Python
364 lines
16 KiB
Python
"""Connected apps: sign-in to external applications with a Bambuddy account.
|
|
|
|
These lock down what a code is worth: single use, 60 seconds, one app, its
|
|
exact callback URL, the PKCE challenge, and only with the app's secret. And
|
|
that none of it works while Bambuddy authentication is disabled.
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
CALLBACK = "http://orders.local:8090/auth/callback"
|
|
|
|
|
|
def _pkce() -> tuple[str, str]:
|
|
verifier = secrets.token_urlsafe(48)
|
|
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
|
|
return verifier, challenge
|
|
|
|
|
|
def _auth(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
async def admin_token(async_client: AsyncClient) -> str:
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={"auth_enabled": True, "admin_username": "connectadmin", "admin_password": "AdminPass1!"},
|
|
)
|
|
login = await async_client.post("/api/v1/auth/login", json={"username": "connectadmin", "password": "AdminPass1!"})
|
|
return login.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture
|
|
async def operator(async_client: AsyncClient, admin_token: str) -> dict:
|
|
groups = (await async_client.get("/api/v1/groups/", headers=_auth(admin_token))).json()
|
|
operators = next(g for g in groups if g["name"] == "Operators")
|
|
user = (
|
|
await async_client.post(
|
|
"/api/v1/users/",
|
|
headers=_auth(admin_token),
|
|
json={"username": "shopworker", "password": "Operatorpass1!", "group_ids": [operators["id"]]},
|
|
)
|
|
).json()
|
|
login = await async_client.post("/api/v1/auth/login", json={"username": "shopworker", "password": "Operatorpass1!"})
|
|
return {"id": user["id"], "token": login.json()["access_token"]}
|
|
|
|
|
|
async def _register(async_client: AsyncClient, admin_token: str, **overrides) -> dict:
|
|
payload = {"name": "Bambuddy Orders", "redirect_uri": CALLBACK, **overrides}
|
|
response = await async_client.post("/api/v1/connect/apps", headers=_auth(admin_token), json=payload)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
async def _authorize(async_client: AsyncClient, user_token: str, app: dict, challenge: str, **overrides):
|
|
payload = {
|
|
"client_id": app["client_id"],
|
|
"redirect_uri": app["redirect_uri"],
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
**overrides,
|
|
}
|
|
return await async_client.post("/api/v1/connect/authorize", headers=_auth(user_token), json=payload)
|
|
|
|
|
|
async def _code(async_client: AsyncClient, user_token: str, app: dict) -> tuple[str, str]:
|
|
verifier, challenge = _pkce()
|
|
response = await _authorize(async_client, user_token, app, challenge)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()["code"], verifier
|
|
|
|
|
|
async def _exchange(async_client: AsyncClient, app: dict, code: str, verifier: str, **overrides):
|
|
payload = {
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": app["redirect_uri"],
|
|
"client_id": app["client_id"],
|
|
"client_secret": app["client_secret"],
|
|
"code_verifier": verifier,
|
|
**overrides,
|
|
}
|
|
# No Authorization header: the app authenticates with its secret alone.
|
|
return await async_client.post("/api/v1/connect/token", json=payload)
|
|
|
|
|
|
class TestRegistration:
|
|
async def test_secret_is_shown_once(self, async_client, admin_token):
|
|
app = await _register(async_client, admin_token)
|
|
assert app["client_id"].startswith("bba_")
|
|
assert app["client_secret"].startswith("bbs_")
|
|
listed = (await async_client.get("/api/v1/connect/apps", headers=_auth(admin_token))).json()
|
|
assert [a["client_id"] for a in listed] == [app["client_id"]]
|
|
assert "client_secret" not in listed[0]
|
|
|
|
async def test_refused_while_authentication_is_disabled(self, async_client):
|
|
response = await async_client.post("/api/v1/connect/apps", json={"name": "Orders", "redirect_uri": CALLBACK})
|
|
assert response.status_code == 400
|
|
|
|
async def test_non_admin_cannot_register(self, async_client, operator):
|
|
response = await async_client.post(
|
|
"/api/v1/connect/apps",
|
|
headers=_auth(operator["token"]),
|
|
json={"name": "Orders", "redirect_uri": CALLBACK},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.parametrize(
|
|
"redirect_uri",
|
|
[
|
|
"javascript:alert(1)",
|
|
"/relative/callback",
|
|
"http://orders.local/cb#frag",
|
|
"http://user:pw@orders.local/cb",
|
|
"ftp://orders.local/cb",
|
|
],
|
|
)
|
|
async def test_callback_must_be_a_plain_absolute_http_url(self, async_client, admin_token, redirect_uri):
|
|
response = await async_client.post(
|
|
"/api/v1/connect/apps",
|
|
headers=_auth(admin_token),
|
|
json={"name": "Orders", "redirect_uri": redirect_uri},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
class TestSignIn:
|
|
async def test_full_flow_returns_identity_and_permissions(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
|
|
info = await async_client.get(
|
|
"/api/v1/connect/authorize/info",
|
|
headers=_auth(operator["token"]),
|
|
params={"client_id": app["client_id"], "redirect_uri": CALLBACK},
|
|
)
|
|
assert info.status_code == 200
|
|
assert info.json() == {"app_name": "Bambuddy Orders", "username": "shopworker", "already_granted": False}
|
|
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
response = await _exchange(async_client, app, code, verifier)
|
|
assert response.status_code == 200, response.text
|
|
user = response.json()["user"]
|
|
assert user["id"] == operator["id"]
|
|
assert user["username"] == "shopworker"
|
|
assert user["is_admin"] is False
|
|
assert user["groups"] == ["Operators"]
|
|
assert "queue:create" in user["permissions"]
|
|
|
|
info_after = await async_client.get(
|
|
"/api/v1/connect/authorize/info",
|
|
headers=_auth(operator["token"]),
|
|
params={"client_id": app["client_id"], "redirect_uri": CALLBACK},
|
|
)
|
|
assert info_after.json()["already_granted"] is True
|
|
|
|
async def test_authorize_needs_a_login_not_an_api_key(self, async_client, admin_token):
|
|
app = await _register(async_client, admin_token)
|
|
key = (
|
|
await async_client.post("/api/v1/api-keys/", headers=_auth(admin_token), json={"name": "script"})
|
|
).json()["key"]
|
|
_, challenge = _pkce()
|
|
for headers in ({"X-API-Key": key}, _auth(key)):
|
|
response = await async_client.post(
|
|
"/api/v1/connect/authorize",
|
|
headers=headers,
|
|
json={
|
|
"client_id": app["client_id"],
|
|
"redirect_uri": CALLBACK,
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
async def test_callback_must_match_exactly_at_authorize(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
_, challenge = _pkce()
|
|
for wrong in (CALLBACK + "/", CALLBACK + "?x=1", "http://evil.example/auth/callback"):
|
|
response = await _authorize(async_client, operator["token"], app, challenge, redirect_uri=wrong)
|
|
assert response.status_code == 400
|
|
|
|
async def test_only_s256_is_accepted(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
verifier, _ = _pkce()
|
|
response = await _authorize(async_client, operator["token"], app, verifier[:43], code_challenge_method="plain")
|
|
assert response.status_code == 422
|
|
|
|
|
|
class TestCodeIsWorthLittle:
|
|
async def test_code_is_single_use(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 200
|
|
replay = await _exchange(async_client, app, code, verifier)
|
|
assert replay.status_code == 400
|
|
assert replay.json()["detail"] == {"error": "invalid_grant"}
|
|
|
|
async def test_expired_code_is_refused(self, async_client, admin_token, operator, db_session):
|
|
from sqlalchemy import update
|
|
|
|
from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
|
|
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
await db_session.execute(
|
|
update(AuthEphemeralToken)
|
|
.where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
|
|
.values(expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))
|
|
)
|
|
await db_session.commit()
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 400
|
|
|
|
async def test_code_is_not_stored_in_plain(self, async_client, admin_token, operator, db_session):
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
|
|
|
|
app = await _register(async_client, admin_token)
|
|
code, _ = await _code(async_client, operator["token"], app)
|
|
stored = (
|
|
await db_session.execute(
|
|
select(AuthEphemeralToken.token).where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
|
|
)
|
|
).scalar_one()
|
|
assert stored != code
|
|
|
|
async def test_wrong_verifier_spends_the_code(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
other_verifier, _ = _pkce()
|
|
assert (await _exchange(async_client, app, code, other_verifier)).status_code == 400
|
|
# A failed attempt burns the code, so the right verifier can't be tried next.
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 400
|
|
|
|
async def test_wrong_callback_at_exchange_is_refused(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
response = await _exchange(async_client, app, code, verifier, redirect_uri=CALLBACK + "x")
|
|
assert response.status_code == 400
|
|
|
|
async def test_wrong_secret_is_invalid_client(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
response = await _exchange(async_client, app, code, verifier, client_secret="bbs_wrong")
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == {"error": "invalid_client"}
|
|
# The code survives a caller that can't prove who it is.
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 200
|
|
|
|
async def test_unknown_client_is_invalid_client(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
response = await _exchange(async_client, app, code, verifier, client_id="bba_unknown")
|
|
assert response.status_code == 401
|
|
|
|
async def test_one_apps_code_is_useless_to_another(self, async_client, admin_token, operator):
|
|
app_a = await _register(async_client, admin_token, name="A")
|
|
app_b = await _register(async_client, admin_token, name="B")
|
|
code, verifier = await _code(async_client, operator["token"], app_a)
|
|
assert (await _exchange(async_client, app_b, code, verifier)).status_code == 400
|
|
|
|
async def test_user_disabled_after_consent_is_refused(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
await async_client.patch(
|
|
f"/api/v1/users/{operator['id']}", headers=_auth(admin_token), json={"is_active": False}
|
|
)
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 400
|
|
|
|
|
|
class TestAppLifecycle:
|
|
async def test_disabled_app_can_neither_authorize_nor_exchange(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
await async_client.patch(
|
|
f"/api/v1/connect/apps/{app['id']}", headers=_auth(admin_token), json={"enabled": False}
|
|
)
|
|
_, challenge = _pkce()
|
|
assert (await _authorize(async_client, operator["token"], app, challenge)).status_code == 400
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 401
|
|
|
|
async def test_rotated_secret_replaces_the_old_one(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
rotated = (
|
|
await async_client.post(f"/api/v1/connect/apps/{app['id']}/rotate-secret", headers=_auth(admin_token))
|
|
).json()
|
|
assert rotated["client_secret"] != app["client_secret"]
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 401
|
|
assert (
|
|
await _exchange(async_client, app, code, verifier, client_secret=rotated["client_secret"])
|
|
).status_code == 200
|
|
|
|
async def test_changed_callback_invalidates_codes_for_the_old_one(self, async_client, admin_token, operator):
|
|
app = await _register(async_client, admin_token)
|
|
code, verifier = await _code(async_client, operator["token"], app)
|
|
await async_client.patch(
|
|
f"/api/v1/connect/apps/{app['id']}",
|
|
headers=_auth(admin_token),
|
|
json={"redirect_uri": "http://orders.local:9000/cb"},
|
|
)
|
|
assert (await _exchange(async_client, app, code, verifier)).status_code == 400
|
|
|
|
async def test_deleting_the_app_removes_grants_and_codes(self, async_client, admin_token, operator, db_session):
|
|
from sqlalchemy import func, select
|
|
|
|
from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
|
|
from backend.app.models.connected_app import ConnectedAppGrant
|
|
|
|
app = await _register(async_client, admin_token)
|
|
await _code(async_client, operator["token"], app)
|
|
response = await async_client.delete(f"/api/v1/connect/apps/{app['id']}", headers=_auth(admin_token))
|
|
assert response.status_code == 200
|
|
grants = await db_session.execute(select(func.count()).select_from(ConnectedAppGrant))
|
|
codes = await db_session.execute(
|
|
select(func.count())
|
|
.select_from(AuthEphemeralToken)
|
|
.where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
|
|
)
|
|
assert grants.scalar_one() == 0
|
|
assert codes.scalar_one() == 0
|
|
|
|
|
|
class TestAuthenticationDisabled:
|
|
async def test_authorize_and_token_say_auth_disabled(self, async_client):
|
|
info = await async_client.get(
|
|
"/api/v1/connect/authorize/info", params={"client_id": "bba_x", "redirect_uri": CALLBACK}
|
|
)
|
|
assert info.status_code == 409
|
|
assert info.json()["detail"] == "auth_disabled"
|
|
verifier, _ = _pkce()
|
|
token = await async_client.post(
|
|
"/api/v1/connect/token",
|
|
json={
|
|
"grant_type": "authorization_code",
|
|
"code": "x",
|
|
"redirect_uri": CALLBACK,
|
|
"client_id": "bba_x",
|
|
"client_secret": "bbs_x",
|
|
"code_verifier": verifier,
|
|
},
|
|
)
|
|
assert token.status_code == 400
|
|
assert token.json()["detail"] == {"error": "auth_disabled"}
|
|
|
|
|
|
class TestRateLimit:
|
|
async def test_repeated_failures_lock_the_client_out(self, async_client, admin_token, operator):
|
|
from backend.app.api.routes.connected_apps import MAX_FAILED_TOKEN_EXCHANGES
|
|
|
|
app = await _register(async_client, admin_token)
|
|
verifier, _ = _pkce()
|
|
for _ in range(MAX_FAILED_TOKEN_EXCHANGES):
|
|
response = await _exchange(async_client, app, "not-a-code", verifier)
|
|
assert response.status_code == 400
|
|
code, good_verifier = await _code(async_client, operator["token"], app)
|
|
assert (await _exchange(async_client, app, code, good_verifier)).status_code == 429
|