Lukas
81f009e323
- Store session tokens as one-way SHA256 hashes instead of plaintext - Hash tokens on write (create_session) and on read (get_session, delete_session) - Add migration to drop plaintext sessions table and recreate with token_hash column - Update Session model: token field still contains raw token for signing - Add test to verify tokens are hashed in database, not plaintext - Update Architekture.md to document session token hashing - Update Backend-Development.md with implementation pattern and best practices Prevents direct session token hijacking if database file is exposed to attacker. If plaintext DB was readable, sessions are invalidated by the migration anyway. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""Tests for app.db — database schema initialisation."""
|
|
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from app.db import init_db
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_settings_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``settings`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='settings';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_sessions_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``sessions`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='sessions';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_blocklist_sources_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``blocklist_sources`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='blocklist_sources';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_import_log_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``import_log`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='import_log';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_is_idempotent(tmp_path: Path) -> None:
|
|
"""Calling ``init_db`` twice on the same database must not raise."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
await init_db(db) # Second call must be a no-op.
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_records_schema_version(tmp_path: Path) -> None:
|
|
"""``init_db`` must record the current schema version."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1;"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
assert row[0] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_migrates_legacy_database_without_schema_version(tmp_path: Path) -> None:
|
|
"""Legacy databases without a schema marker must be migrated on startup."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
await db.execute("DROP TABLE schema_migrations;")
|
|
await db.commit()
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1;"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
assert row[0] == 2
|