Addressed SA-3 by encrypting the sqlite content with a .env secret

backend/app/api/admin.py

@@ -20,6 +20,7 @@ from backend.app.services.email_service import (
 )
 from backend.app.services.email_verification import create_verification_token
 from backend.app.services.theme_service import THEMES, get_enabled_themes, save_enabled_themes
+from backend.app.services.secret_store import encrypt_secret
 from backend.app.core.config import settings
 
 router = APIRouter()
@@ -345,6 +346,9 @@ def update_plugin(
         config = json.loads(current['config']) if current['config'] else {}
         if payload.config is not None:
             config.update(payload.config)
+        for secret_name in ('access_token', 'client_secret', 'smtp_password'):
+            if config.get(secret_name):
+                config[secret_name] = encrypt_secret(config[secret_name])
 
         conn.execute(
             '''

backend/app/api/auth.py

@@ -15,6 +15,7 @@ from backend.app.services.email_verification import verify_email
 from backend.app.services.password_reset import create_reset_token, reset_password
 from backend.app.services.token_service import issue_token, revoke_token, validate_token
 from backend.app.services.otp_service import verify_code
+from backend.app.services.secret_store import decrypt_secret
 from backend.app.services.email_addresses import verify_user_email_address
 
 router = APIRouter()
@@ -48,7 +49,7 @@ def login(payload: LoginRequest):
         raise HTTPException(status_code=401, detail='Invalid username or password')
     if not user['email_verified']:
         raise HTTPException(status_code=403, detail='Email address is not verified')
-    if user['otp_enabled'] and not verify_code(user['otp_secret'], payload.otp):
+    if user['otp_enabled'] and not verify_code(decrypt_secret(user['otp_secret']), payload.otp):
         raise HTTPException(status_code=401, detail='One-time password required or invalid')
 
     token_data = issue_token(user['id'], user['username'])

backend/app/api/user_config.py

@@ -15,6 +15,7 @@ from backend.app.services.otp_service import create_secret, provisioning_uri, ve
 from backend.app.services.email_addresses import add_user_email_address, create_email_verification, list_user_email_addresses
 from backend.app.services.email_service import send_verification_email, smtp_configured
 from backend.app.core.config import settings
+from backend.app.services.secret_store import decrypt_secret, encrypt_secret
 
 router = APIRouter()
 
@@ -116,7 +117,7 @@ def setup_otp(user: dict = Depends(get_current_user)):
         raise HTTPException(status_code=409, detail='One-time password is already enabled')
     secret = create_secret()
     with get_connection() as conn:
-        conn.execute('UPDATE users SET otp_secret = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (secret, user['id']))
+        conn.execute('UPDATE users SET otp_secret = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (encrypt_secret(secret), user['id']))
         conn.commit()
     return {'secret': secret, 'otpauth_url': provisioning_uri(secret, user['username'])}
 
@@ -125,7 +126,7 @@ def setup_otp(user: dict = Depends(get_current_user)):
 def update_otp(payload: OtpUpdate, user: dict = Depends(get_current_user)):
     if payload.action not in {'enable', 'disable'}:
         raise HTTPException(status_code=422, detail='OTP action must be enable or disable')
-    if not verify_code(user['otp_secret'], payload.code):
+    if not verify_code(decrypt_secret(user['otp_secret']), payload.code):
         raise HTTPException(status_code=400, detail='Invalid one-time password')
     with get_connection() as conn:
         if payload.action == 'enable':
@@ -329,6 +330,8 @@ def get_user_plugin_config(plugin_name: str, user: dict = Depends(get_current_us
         return {}
 
     config = json.loads(row['config']) if row['config'] else {}
+    if config.get('access_token'):
+        config.pop('access_token')
     return config
 
 
@@ -346,6 +349,8 @@ def update_user_plugin_config(
 
         current_config = json.loads(current['config']) if current and current['config'] else {}
         updates = payload.model_dump(exclude_none=True)
+        if updates.get('access_token'):
+            updates['access_token'] = encrypt_secret(updates['access_token'])
         merged = {**current_config, **updates}
 
         if current is None:
@@ -368,4 +373,6 @@ def update_user_plugin_config(
 
         conn.commit()
 
-    return merged
+    public_config = dict(merged)
+    public_config.pop('access_token', None)
+    return public_config

backend/app/core/config.py

@@ -24,6 +24,7 @@ class Settings:
     version: str = os.getenv('LINKLOG_VERSION', '0.1.0')
     database_url: str = os.getenv('LINKLOG_DATABASE_URL', f'sqlite:///{DB_PATH}')
     secret_key: str = os.getenv('LINKLOG_SECRET_KEY', 'dev-secret-key-change-me')
+    data_encryption_key: str = os.getenv('LINKLOG_DATA_ENCRYPTION_KEY', '')
     token_expiry_days: int = int(os.getenv('LINKLOG_TOKEN_EXPIRY_DAYS', '30'))
     public_url: str = normalize_public_url(os.getenv('LINKLOG_PUBLIC_URL', 'http://localhost:8000'))
     smtp_host: str = os.getenv('LINKLOG_SMTP_HOST', '')

backend/app/services/email_service.py

@@ -7,6 +7,7 @@ import json
 
 from backend.app.core.config import settings
 from backend.app.database import get_connection
+from backend.app.services.secret_store import decrypt_secret, encrypt_secret
 
 
 def get_smtp_settings() -> dict:
@@ -22,6 +23,7 @@ def get_smtp_settings() -> dict:
         row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('smtp',)).fetchone()
     if row:
         values.update(json.loads(row['value']))
+    values['smtp_password'] = decrypt_secret(values['smtp_password'])
     return values
 
 
@@ -30,7 +32,7 @@ def save_smtp_settings(values: dict) -> None:
         conn.execute(
             '''INSERT INTO app_settings (name, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
                ON CONFLICT(name) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP''',
-            ('smtp', json.dumps(values)),
+            ('smtp', json.dumps({**values, 'smtp_password': encrypt_secret(values['smtp_password'])})),
         )
         conn.commit()
 

backend/app/services/mastodon_oauth.py

@@ -12,6 +12,7 @@ from uuid import uuid4
 
 from backend.app.core.config import settings
 from backend.app.database import get_connection
+from backend.app.services.secret_store import decrypt_secret, encrypt_secret
 
 
 def normalize_instance(instance: str) -> str:
@@ -39,6 +40,8 @@ def start_authorization(user_id: str, instance: str) -> str:
     with get_connection() as conn:
         row = conn.execute('SELECT value FROM app_settings WHERE name = ?', (setting_name,)).fetchone()
     app = json.loads(row['value']) if row else None
+    if app and app.get('client_secret'):
+        app['client_secret'] = decrypt_secret(app['client_secret'])
     if not app:
         app = post_form(f'{instance}/api/v1/apps', {
             'client_name': settings.mastodon_client_name,
@@ -50,7 +53,7 @@ def start_authorization(user_id: str, instance: str) -> str:
             conn.execute(
                 '''INSERT INTO app_settings (name, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
                    ON CONFLICT(name) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP''',
-                (setting_name, json.dumps(app)),
+                (setting_name, json.dumps({**app, 'client_secret': encrypt_secret(app['client_secret'])})),
             )
             conn.commit()
     state = token_urlsafe(32)
@@ -62,7 +65,7 @@ def start_authorization(user_id: str, instance: str) -> str:
                (id, user_id, state_hash, instance, client_id, client_secret, redirect_uri, expires_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)''',
             (str(uuid4()), user_id, sha256(state.encode()).hexdigest(), instance,
-             app['client_id'], app['client_secret'], redirect_uri, expires_at.isoformat()),
+             app['client_id'], encrypt_secret(app['client_secret']), redirect_uri, expires_at.isoformat()),
         )
         conn.commit()
     return f'{instance}/oauth/authorize?' + urlencode({
@@ -90,7 +93,7 @@ def finish_authorization(code: str, state: str) -> str:
         'grant_type': 'authorization_code',
         'code': code,
         'client_id': record['client_id'],
-        'client_secret': record['client_secret'],
+        'client_secret': decrypt_secret(record['client_secret']),
         'redirect_uri': record['redirect_uri'],
     })
     access_token = token.get('access_token')
@@ -102,7 +105,7 @@ def finish_authorization(code: str, state: str) -> str:
             (record['user_id'], 'mastodon'),
         ).fetchone()
         config = json.loads(current['config']) if current and current['config'] else {}
-        config.update({'instance': record['instance'], 'access_token': access_token})
+        config.update({'instance': record['instance'], 'access_token': encrypt_secret(access_token)})
         if current:
             conn.execute(
                 'UPDATE user_plugin_config SET config = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND plugin_name = ?',

backend/app/services/plugin_manager.py

@@ -8,6 +8,7 @@ from urllib.parse import urlencode
 from urllib.request import Request, urlopen
 
 from backend.app.plugins.base import BasePlugin
+from backend.app.services.secret_store import decrypt_secret
 
 DEFAULT_POST_PREFIX = 'From my #LinkLog: '
 logger = logging.getLogger(__name__)
@@ -47,6 +48,7 @@ class MastodonPlugin(BasePlugin):
                 ).fetchone()
             if row and row['config']:
                 config.update(json.loads(row['config']))
+        config['access_token'] = decrypt_secret(config.get('access_token', ''))
 
         instance = str(config.get('instance', '')).strip().rstrip('/')
         if instance and '://' not in instance:
@@ -135,6 +137,7 @@ class MastodonPlugin(BasePlugin):
                 ).fetchone()
             if row and row['config']:
                 config.update(json.loads(row['config']))
+        config['access_token'] = decrypt_secret(config.get('access_token', ''))
 
         instance = str(config.get('instance', '')).strip().rstrip('/')
         if instance and '://' not in instance:

backend/app/services/secret_store.py

@@ -0,0 +1,32 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+from cryptography.fernet import Fernet, InvalidToken
+
+from backend.app.core.config import settings
+
+
+def _cipher() -> Fernet:
+    if not settings.data_encryption_key:
+        raise RuntimeError('LINKLOG_DATA_ENCRYPTION_KEY is required to access encrypted secrets')
+    try:
+        return Fernet(settings.data_encryption_key.encode('ascii'))
+    except (ValueError, UnicodeEncodeError) as error:
+        raise RuntimeError('LINKLOG_DATA_ENCRYPTION_KEY must be a valid Fernet key') from error
+
+
+def encrypt_secret(value: str) -> str:
+    if not value:
+        return value
+    if value.startswith('enc:v1:'):
+        return value
+    return 'enc:v1:' + _cipher().encrypt(value.encode('utf-8')).decode('ascii')
+
+
+def decrypt_secret(value: str) -> str:
+    if not value or not value.startswith('enc:v1:'):
+        return value
+    try:
+        return _cipher().decrypt(value[7:].encode('ascii')).decode('utf-8')
+    except (InvalidToken, UnicodeEncodeError) as error:
+        raise RuntimeError('Encrypted secret cannot be decrypted with LINKLOG_DATA_ENCRYPTION_KEY') from error

backend/requirements.txt

@@ -6,3 +6,4 @@ python-multipart==0.0.20
 pytest==9.1.1
 httpx==0.28.1
 httpx2==2.12.0
+cryptography==46.0.3

backend/tests/conftest.py

@@ -10,6 +10,7 @@ import pytest
 TEST_DATABASE_DIRECTORY = tempfile.TemporaryDirectory(prefix='linklog-tests-')
 TEST_DATABASE_PATH = os.path.join(TEST_DATABASE_DIRECTORY.name, 'linklog.db')
 os.environ['LINKLOG_DATABASE_PATH'] = TEST_DATABASE_PATH
+os.environ['LINKLOG_DATA_ENCRYPTION_KEY'] = 'L5M4sQYVjD1N7pT2Xk8R0aBcDeFgHiJkLmNoPqRsTuV='
 
 
 @pytest.fixture(scope='session', autouse=True)

backend/tests/test_email_service.py

@@ -39,4 +39,19 @@ def test_send_test_email_uses_configured_recipient(monkeypatch):
 
     message = smtp.send_message.call_args.args[0]
     assert message['To'] == 'admin@example.com'
-    assert message['Subject'] == 'LinkLog SMTP test'
+    assert message['Subject'] == 'LinkLog SMTP test'
+
+
+def test_smtp_password_is_encrypted_at_rest():
+    from backend.app.services.email_service import get_smtp_settings, save_smtp_settings
+    from backend.app.database import get_connection
+
+    values = {
+        'smtp_host': 'smtp.example.com', 'smtp_port': 587, 'smtp_username': 'mailer',
+        'smtp_password': 'secret', 'smtp_from': 'LinkLog <no-reply@example.com>', 'smtp_use_tls': True,
+    }
+    save_smtp_settings(values)
+    with get_connection() as conn:
+        stored = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('smtp',)).fetchone()['value']
+    assert 'secret' not in stored
+    assert get_smtp_settings()['smtp_password'] == 'secret'