mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +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.
335 lines
13 KiB
Python
335 lines
13 KiB
Python
"""Integration tests for OIDC/MFA cleanup on user deletion.
|
|
|
|
These tests verify the fix for issue #1285: deleting a user via DELETE
|
|
/api/v1/users/{id} must also remove their UserOIDCLink, UserTOTP, and
|
|
UserOTPCode rows. On PostgreSQL the FK CASCADE handles this, but SQLite
|
|
ships with FK enforcement off — without explicit DELETEs in the endpoint,
|
|
orphan rows would block SSO re-login and leak MFA secrets.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
class TestDeleteUserCleansAuthRows:
|
|
"""Verify delete_user removes OIDC link + TOTP + OTP rows owned by the user."""
|
|
|
|
@pytest.fixture
|
|
async def auth_token(self, async_client: AsyncClient):
|
|
"""Setup auth and return admin token."""
|
|
await async_client.post(
|
|
"/api/v1/auth/setup",
|
|
json={
|
|
"auth_enabled": True,
|
|
"admin_username": "cleanupadmin",
|
|
"admin_password": "AdminPass1!",
|
|
},
|
|
)
|
|
login_response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "cleanupadmin", "password": "AdminPass1!"},
|
|
)
|
|
return login_response.json()["access_token"]
|
|
|
|
async def _create_user(self, async_client: AsyncClient, auth_token: str, username: str) -> int:
|
|
"""Helper: create a non-admin user via the API and return their id."""
|
|
create_resp = await async_client.post(
|
|
"/api/v1/users/",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
json={
|
|
"username": username,
|
|
"password": "Password123!",
|
|
"role": "user",
|
|
},
|
|
)
|
|
assert create_resp.status_code in (200, 201), create_resp.text
|
|
return create_resp.json()["id"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_removes_oidc_links(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""Deleting a user must also delete their UserOIDCLink rows."""
|
|
from backend.app.models.oidc_provider import OIDCProvider, UserOIDCLink
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "oidcclean")
|
|
|
|
# Use the client_secret property setter (mfa_encrypt) instead of poking
|
|
# _client_secret_enc directly — keeps the fixture in sync with the real
|
|
# encryption flow even though nothing decrypts it in this test
|
|
# (#1295 review nit).
|
|
provider = OIDCProvider(
|
|
name="CleanupProv",
|
|
issuer_url="https://cleanup.example.com",
|
|
client_id="cleanup_client",
|
|
scopes="openid email profile",
|
|
is_enabled=True,
|
|
)
|
|
provider.client_secret = "cleanup_secret"
|
|
db_session.add(provider)
|
|
await db_session.flush()
|
|
db_session.add(
|
|
UserOIDCLink(
|
|
user_id=user_id,
|
|
provider_id=provider.id,
|
|
provider_user_id="sub-cleanup-123",
|
|
provider_email="cleanup@example.com",
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
# Sanity check: link exists before delete
|
|
pre = await db_session.execute(select(UserOIDCLink).where(UserOIDCLink.user_id == user_id))
|
|
assert pre.scalar_one_or_none() is not None
|
|
|
|
# Delete via API
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Link must be gone (the bug from #1285 is when it persists on SQLite)
|
|
await db_session.commit()
|
|
post = await db_session.execute(select(UserOIDCLink).where(UserOIDCLink.user_id == user_id))
|
|
assert post.scalar_one_or_none() is None, "UserOIDCLink orphan left behind — #1285 regression"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_removes_user_totp(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""Deleting a user must also delete their UserTOTP row (MFA secret)."""
|
|
from backend.app.models.user_totp import UserTOTP
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "totpclean")
|
|
|
|
totp = UserTOTP(user_id=user_id, is_enabled=True)
|
|
totp.secret = "JBSWY3DPEHPK3PXP" # encrypts via property setter
|
|
db_session.add(totp)
|
|
await db_session.commit()
|
|
|
|
pre = await db_session.execute(select(UserTOTP).where(UserTOTP.user_id == user_id))
|
|
assert pre.scalar_one_or_none() is not None
|
|
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
await db_session.commit()
|
|
post = await db_session.execute(select(UserTOTP).where(UserTOTP.user_id == user_id))
|
|
assert post.scalar_one_or_none() is None, "UserTOTP orphan — MFA secret leaked after user delete"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_removes_long_lived_tokens(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""Deleting a user must also delete their LongLivedToken rows.
|
|
|
|
Camera-stream tokens whose `secret_hash` is still valid would
|
|
otherwise be matchable by `verify()` via `lookup_prefix` even
|
|
after the user is gone (#1295 review feedback).
|
|
"""
|
|
from backend.app.models.long_lived_token import LongLivedToken
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "lltclean")
|
|
|
|
db_session.add(
|
|
LongLivedToken(
|
|
user_id=user_id,
|
|
name="HA card",
|
|
lookup_prefix="abcd1234",
|
|
secret_hash="$2b$12$dummybcrypthashabcdefghij1234567890",
|
|
scope="camera_stream",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(days=30),
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
pre = await db_session.execute(select(LongLivedToken).where(LongLivedToken.user_id == user_id))
|
|
assert pre.scalar_one_or_none() is not None
|
|
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
await db_session.commit()
|
|
post = await db_session.execute(select(LongLivedToken).where(LongLivedToken.user_id == user_id))
|
|
assert post.scalar_one_or_none() is None, (
|
|
"LongLivedToken orphan — camera-stream secret still in DB after user delete"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_removes_connected_app_consent_and_codes(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""A leftover consent row would skip the consent screen for the next
|
|
user given the same id; a live code is keyed by username."""
|
|
from backend.app.models.auth_ephemeral import AuthEphemeralToken, TokenType
|
|
from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "appclean")
|
|
app = ConnectedApp(
|
|
name="Orders", client_id="bba_cleanup", client_secret_hash="x", redirect_uri="http://o.local/cb"
|
|
)
|
|
db_session.add(app)
|
|
await db_session.flush()
|
|
db_session.add(ConnectedAppGrant(app_id=app.id, user_id=user_id))
|
|
db_session.add(
|
|
AuthEphemeralToken.new_connect_code(
|
|
code_hash="h" * 64,
|
|
username="AppClean",
|
|
app_id=app.id,
|
|
code_challenge="c" * 43,
|
|
expires_at=datetime.now(timezone.utc) + timedelta(seconds=60),
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
await db_session.commit()
|
|
grants = await db_session.execute(select(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id))
|
|
assert grants.scalar_one_or_none() is None
|
|
codes = await db_session.execute(
|
|
select(AuthEphemeralToken).where(AuthEphemeralToken.token_type == TokenType.CONNECT_CODE)
|
|
)
|
|
assert codes.scalar_one_or_none() is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_removes_user_otp_codes(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""Deleting a user must also delete their UserOTPCode rows."""
|
|
from backend.app.models.user_otp_code import UserOTPCode
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "otpclean")
|
|
|
|
# Two pending OTP codes so we verify the WHERE clause hits all rows
|
|
for _ in range(2):
|
|
db_session.add(
|
|
UserOTPCode(
|
|
user_id=user_id,
|
|
code_hash="$pbkdf2-sha256$dummy",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(minutes=10),
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
pre = await db_session.execute(select(UserOTPCode).where(UserOTPCode.user_id == user_id))
|
|
assert len(pre.scalars().all()) == 2
|
|
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
await db_session.commit()
|
|
post = await db_session.execute(select(UserOTPCode).where(UserOTPCode.user_id == user_id))
|
|
assert post.scalars().all() == [], "UserOTPCode orphans left behind"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_user_with_all_auth_rows(
|
|
self,
|
|
async_client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
auth_token: str,
|
|
):
|
|
"""Combined: one user with OIDC link + TOTP + OTP + long-lived token — all cleaned up atomically."""
|
|
from backend.app.models.long_lived_token import LongLivedToken
|
|
from backend.app.models.oidc_provider import OIDCProvider, UserOIDCLink
|
|
from backend.app.models.user_otp_code import UserOTPCode
|
|
from backend.app.models.user_totp import UserTOTP
|
|
|
|
user_id = await self._create_user(async_client, auth_token, "fullauth")
|
|
|
|
provider = OIDCProvider(
|
|
name="FullAuthProv",
|
|
issuer_url="https://fullauth.example.com",
|
|
client_id="fullauth_client",
|
|
scopes="openid email profile",
|
|
is_enabled=True,
|
|
)
|
|
provider.client_secret = "fullauth_secret"
|
|
db_session.add(provider)
|
|
await db_session.flush()
|
|
|
|
db_session.add(
|
|
UserOIDCLink(
|
|
user_id=user_id,
|
|
provider_id=provider.id,
|
|
provider_user_id="sub-fullauth",
|
|
provider_email="full@example.com",
|
|
)
|
|
)
|
|
totp = UserTOTP(user_id=user_id, is_enabled=True)
|
|
totp.secret = "JBSWY3DPEHPK3PXP"
|
|
db_session.add(totp)
|
|
db_session.add(
|
|
UserOTPCode(
|
|
user_id=user_id,
|
|
code_hash="$pbkdf2-sha256$dummy",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(minutes=10),
|
|
)
|
|
)
|
|
db_session.add(
|
|
LongLivedToken(
|
|
user_id=user_id,
|
|
name="combined-test",
|
|
lookup_prefix="zz999999",
|
|
secret_hash="$2b$12$dummybcrypthashabcdefghij1234567890",
|
|
scope="camera_stream",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(days=30),
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.delete(
|
|
f"/api/v1/users/{user_id}",
|
|
headers={"Authorization": f"Bearer {auth_token}"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
await db_session.commit()
|
|
link_post = await db_session.execute(select(UserOIDCLink).where(UserOIDCLink.user_id == user_id))
|
|
totp_post = await db_session.execute(select(UserTOTP).where(UserTOTP.user_id == user_id))
|
|
otp_post = await db_session.execute(select(UserOTPCode).where(UserOTPCode.user_id == user_id))
|
|
llt_post = await db_session.execute(select(LongLivedToken).where(LongLivedToken.user_id == user_id))
|
|
assert link_post.scalar_one_or_none() is None
|
|
assert totp_post.scalar_one_or_none() is None
|
|
assert otp_post.scalars().all() == []
|
|
assert llt_post.scalar_one_or_none() is None
|