Initial config with email service test

backend/app/api/auth.py

@@ -7,8 +7,11 @@ from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
 from backend.app.database import get_connection, init_db
-from backend.app.services.auth_service import authenticate_user
+from backend.app.core.config import settings
+from backend.app.services.auth_service import authenticate_user, find_user
+from backend.app.services.email_service import send_password_reset_email, smtp_configured
 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
 
 router = APIRouter()
@@ -21,10 +24,23 @@ class LoginRequest(BaseModel):
     password: str
 
 
+class PasswordResetRequest(BaseModel):
+    token: str
+    password: str
+
+
 @router.post('/login')
 def login(payload: LoginRequest):
     user = authenticate_user(payload.username, payload.password)
     if user is None:
+        reset_user = find_user(payload.username)
+        if reset_user and reset_user['email_verified'] and smtp_configured():
+            try:
+                token = create_reset_token(reset_user['id'])
+                reset_url = f'{settings.public_url}/reset-password?token={token}'
+                send_password_reset_email(reset_user['email'], reset_user['username'], reset_url)
+            except Exception:
+                pass
         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')
@@ -46,6 +62,15 @@ def verify_email_address(token: str):
     return {'status': 'verified', 'message': 'Email address verified. You can now sign in.'}
 
 
+@router.post('/reset-password')
+def reset_password_endpoint(payload: PasswordResetRequest):
+    if len(payload.password) < 8:
+        raise HTTPException(status_code=422, detail='Password must contain at least 8 characters')
+    if not reset_password(payload.token, payload.password):
+        raise HTTPException(status_code=400, detail='Password reset link is invalid or expired')
+    return {'status': 'password_reset', 'message': 'Password reset. You can now sign in.'}
+
+
 @router.post('/logout')
 def logout(payload: dict):
     token = payload.get('token')

backend/app/api/setup.py

@@ -1,6 +1,8 @@
 ## Copyright © 2026 Olaf Kolkman
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
+from datetime import datetime, timedelta, timezone
+import json
 from uuid import uuid4
 
 from fastapi import APIRouter, HTTPException
@@ -8,7 +10,7 @@ from pydantic import BaseModel
 
 from backend.app.core.config import settings
 from backend.app.database import get_connection, hash_password
-from backend.app.services.email_service import save_smtp_settings, send_test_email
+from backend.app.services.email_service import get_smtp_settings, save_smtp_settings, send_test_email
 
 router = APIRouter()
 
@@ -25,13 +27,41 @@ class SetupRequest(BaseModel):
     smtp_use_tls: bool = True
 
 
+class TestMailRequest(BaseModel):
+    email: str | None = None
+
+
 def has_administrator() -> bool:
     with get_connection() as conn:
         return conn.execute('SELECT 1 FROM users WHERE is_admin = 1 LIMIT 1').fetchone() is not None
 
 
-@router.post('', status_code=201)
-def configure_application(payload: SetupRequest):
+def get_pending_setup() -> dict | None:
+    with get_connection() as conn:
+        row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('setup_pending',)).fetchone()
+    return json.loads(row['value']) if row else None
+
+
+def save_pending_setup(values: dict) -> None:
+    with get_connection() as conn:
+        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''',
+            ('setup_pending', json.dumps(values)),
+        )
+        conn.execute('DELETE FROM app_settings WHERE name = ?', ('setup_mail_rate',))
+        conn.commit()
+
+
+def delete_pending_setup() -> None:
+    with get_connection() as conn:
+        conn.execute('DELETE FROM app_settings WHERE name = ?', ('setup_pending',))
+        conn.execute('DELETE FROM app_settings WHERE name = ?', ('setup_mail_rate',))
+        conn.commit()
+
+
+@router.post('/configuration', status_code=200)
+def save_configuration(payload: SetupRequest):
     if has_administrator():
         raise HTTPException(status_code=409, detail='LinkLog is already configured')
     username = payload.username.strip()
@@ -43,7 +73,6 @@ def configure_application(payload: SetupRequest):
     if not 1 <= payload.smtp_port <= 65535:
         raise HTTPException(status_code=422, detail='SMTP port must be between 1 and 65535')
 
-    user_id = str(uuid4())
     smtp_values = {
         'smtp_host': smtp_host,
         'smtp_port': payload.smtp_port,
@@ -52,29 +81,111 @@ def configure_application(payload: SetupRequest):
         'smtp_from': smtp_from,
         'smtp_use_tls': payload.smtp_use_tls,
     }
+    pending = {
+        'username': username,
+        'email': email,
+        'password_hash': hash_password(payload.password),
+    }
+    save_pending_setup(pending)
+    save_smtp_settings(smtp_values)
+    return {'status': 'saved', 'message': 'Configuration saved. Send a test mail to verify SMTP delivery.'}
+
+
+@router.post('/test-mail')
+def test_mail(payload: TestMailRequest | None = None):
+    if has_administrator():
+        raise HTTPException(status_code=409, detail='LinkLog is already configured')
+    pending = get_pending_setup()
+    if pending is None:
+        raise HTTPException(status_code=400, detail='Save the configuration before sending test mail')
+
+    now = datetime.now(timezone.utc)
+    with get_connection() as conn:
+        row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('setup_mail_rate',)).fetchone()
+        rate = json.loads(row['value']) if row else {}
+    last_sent = datetime.fromisoformat(rate['last_sent']) if rate.get('last_sent') else None
+    cooldown_until = datetime.fromisoformat(rate['cooldown_until']) if rate.get('cooldown_until') else None
+    if cooldown_until and now >= cooldown_until:
+        rate = {}
+        last_sent = None
+        cooldown_until = None
+    if cooldown_until and now < cooldown_until:
+        retry_after = int((cooldown_until - now).total_seconds()) + 1
+        raise HTTPException(status_code=429, detail=f'Test mail limit reached. Try again in {retry_after} seconds.', headers={'Retry-After': str(retry_after)})
+    if last_sent and now - last_sent < timedelta(seconds=20):
+        retry_after = int((timedelta(seconds=20) - (now - last_sent)).total_seconds()) + 1
+        raise HTTPException(status_code=429, detail=f'Please wait {retry_after} seconds before sending another test mail.', headers={'Retry-After': str(retry_after)})
+
+    try:
+        send_test_email(payload.email if payload and payload.email else pending['email'])
+    except Exception as error:
+        raise HTTPException(status_code=503, detail=f'SMTP test mail could not be sent: {error}') from error
+
+    sends = int(rate.get('sends', 0)) + 1
+    updated_rate = {'sends': sends, 'last_sent': now.isoformat()}
+    if sends >= 5:
+        updated_rate['cooldown_until'] = (now + timedelta(minutes=2)).isoformat()
+    with get_connection() as conn:
+        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''',
+            ('setup_mail_rate', json.dumps(updated_rate)),
+        )
+        conn.commit()
+    next_allowed = datetime.fromisoformat(updated_rate.get('cooldown_until')) if sends >= 5 else now + timedelta(seconds=20)
+    return {
+        'status': 'sent',
+        'message': 'SMTP test mail sent.',
+        'sends_remaining': max(0, 5 - sends),
+        'cooldown_seconds': 120 if sends >= 5 else 0,
+        'next_allowed_at': next_allowed.isoformat(),
+    }
+
+
+@router.post('/complete', status_code=201)
+def complete_setup():
+    if has_administrator():
+        raise HTTPException(status_code=409, detail='LinkLog is already configured')
+    pending = get_pending_setup()
+    if pending is None:
+        raise HTTPException(status_code=400, detail='Save the configuration before completing setup')
+    with get_connection() as conn:
+        row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('setup_mail_rate',)).fetchone()
+    if row is None or int(json.loads(row['value']).get('sends', 0)) < 1:
+        raise HTTPException(status_code=400, detail='Send a successful test mail before completing setup')
+    user_id = str(uuid4())
     with get_connection() as conn:
         try:
             conn.execute(
                 '''INSERT INTO users
                    (id, username, email, password_hash, is_admin, email_verified)
                    VALUES (?, ?, ?, ?, 1, 1)''',
-                (user_id, username, email, hash_password(payload.password)),
+                (user_id, pending['username'], pending['email'], pending['password_hash']),
             )
+            conn.commit()
         except Exception as error:
             raise HTTPException(status_code=409, detail='Username or email already exists') from error
-        conn.commit()
-    save_smtp_settings(smtp_values)
-    try:
-        send_test_email(email)
-    except Exception as error:
-        with get_connection() as conn:
-            conn.execute('DELETE FROM users WHERE id = ?', (user_id,))
-            conn.execute('DELETE FROM app_settings WHERE name = ?', ('smtp',))
-            conn.commit()
-        raise HTTPException(status_code=503, detail=f'SMTP test mail could not be sent: {error}') from error
-    return {'status': 'configured', 'message': 'LinkLog is configured and the SMTP test mail was sent.'}
+    delete_pending_setup()
+    return {'status': 'configured', 'message': 'LinkLog is configured.'}
 
 
 @router.get('/status')
 def setup_status():
-    return {'configured': has_administrator()}
+    rate = {}
+    with get_connection() as conn:
+        row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('setup_mail_rate',)).fetchone()
+    if row:
+        rate = json.loads(row['value'])
+    now = datetime.now(timezone.utc)
+    cooldown_until = datetime.fromisoformat(rate['cooldown_until']) if rate.get('cooldown_until') else None
+    last_sent = datetime.fromisoformat(rate['last_sent']) if rate.get('last_sent') else None
+    next_allowed = cooldown_until if cooldown_until and cooldown_until > now else (
+        last_sent + timedelta(seconds=20) if last_sent else None
+    )
+    return {
+        'configured': has_administrator(),
+        'pending': get_pending_setup() is not None,
+        'smtp_defaults': get_smtp_settings(),
+        'sends_remaining': max(0, 5 - int(rate.get('sends', 0))),
+        'next_allowed_at': next_allowed.isoformat() if next_allowed and next_allowed > now else None,
+    }

backend/app/core/config.py

@@ -25,6 +25,7 @@ class Settings:
     smtp_from: str = os.getenv('LINKLOG_SMTP_FROM', 'LinkLog <no-reply@localhost>')
     smtp_use_tls: bool = os.getenv('LINKLOG_SMTP_USE_TLS', 'true').lower() in {'1', 'true', 'yes'}
     email_verification_expiry_hours: int = int(os.getenv('LINKLOG_EMAIL_VERIFICATION_EXPIRY_HOURS', '24'))
+    password_reset_expiry_hours: int = int(os.getenv('LINKLOG_PASSWORD_RESET_EXPIRY_HOURS', '1'))
     tracking_params: list[str] = None
 
     def __post_init__(self):

backend/app/database.py

@@ -123,6 +123,18 @@ CREATE TABLE IF NOT EXISTS app_settings (
     value TEXT NOT NULL,
     updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
+'''),
+    (7, '''
+CREATE TABLE IF NOT EXISTS password_reset_tokens (
+    id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    token_hash TEXT NOT NULL UNIQUE,
+    expires_at TEXT NOT NULL,
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE
+);
+CREATE INDEX IF NOT EXISTS idx_password_reset_tokens_user_id
+    ON password_reset_tokens(user_id);
 '''),
 ]
 

backend/app/main.py

@@ -73,6 +73,11 @@ async def login_page(request: Request):
     return templates.TemplateResponse(request, 'login.html', {})
 
 
+@app.get('/reset-password', response_class=HTMLResponse)
+async def reset_password_page(request: Request):
+    return templates.TemplateResponse(request, 'reset-password.html', {})
+
+
 @app.get('/setup', response_class=HTMLResponse)
 async def setup_page(request: Request):
     if has_administrator():

backend/app/services/auth_service.py

@@ -19,3 +19,9 @@ def authenticate_user(username: str, password: str):
             (username, password_hash),
         ).fetchone()
     return dict(row) if row else None
+
+
+def find_user(username: str):
+    with get_connection() as conn:
+        row = conn.execute('SELECT * FROM users WHERE username = ?', (username,)).fetchone()
+    return dict(row) if row else None

backend/app/services/email_service.py

@@ -70,4 +70,14 @@ def send_verification_email(email: str, username: str, verification_url: str) ->
 
 
 def send_test_email(email: str) -> None:
-    send_message(email, 'LinkLog SMTP test', 'This is a test message from LinkLog. SMTP is configured correctly.\n')
+    send_message(email, 'LinkLog SMTP test', 'This is a test message from LinkLog. SMTP is configured correctly.\n')
+
+
+def send_password_reset_email(email: str, username: str, reset_url: str) -> None:
+    send_message(
+        email,
+        'Reset your LinkLog password',
+        f'Hello {username},\n\n'
+        f'Reset your LinkLog password by opening this link:\n{reset_url}\n\n'
+        f'This link expires in {settings.password_reset_expiry_hours} hours.\n',
+    )

backend/app/services/password_reset.py

@@ -0,0 +1,48 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+from datetime import datetime, timedelta, timezone
+from hashlib import sha256
+from secrets import token_urlsafe
+from uuid import uuid4
+
+from backend.app.core.config import settings
+from backend.app.database import get_connection, hash_password
+
+
+def hash_reset_token(token: str) -> str:
+    return sha256(token.encode('utf-8')).hexdigest()
+
+
+def create_reset_token(user_id: str) -> str:
+    token = token_urlsafe(32)
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=settings.password_reset_expiry_hours)
+    with get_connection() as conn:
+        conn.execute('DELETE FROM password_reset_tokens WHERE user_id = ?', (user_id,))
+        conn.execute(
+            '''INSERT INTO password_reset_tokens
+               (id, user_id, token_hash, expires_at) VALUES (?, ?, ?, ?)''',
+            (str(uuid4()), user_id, hash_reset_token(token), expires_at.isoformat()),
+        )
+        conn.commit()
+    return token
+
+
+def reset_password(token: str, password: str) -> bool:
+    now = datetime.now(timezone.utc).isoformat()
+    with get_connection() as conn:
+        row = conn.execute(
+            '''SELECT user_id FROM password_reset_tokens
+               WHERE token_hash = ? AND expires_at > ?''',
+            (hash_reset_token(token), now),
+        ).fetchone()
+        if row is None:
+            return False
+        conn.execute(
+            'UPDATE users SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
+            (hash_password(password), row['user_id']),
+        )
+        conn.execute('DELETE FROM password_reset_tokens WHERE user_id = ?', (row['user_id'],))
+        conn.execute('DELETE FROM tokens WHERE user_id = ?', (row['user_id'],))
+        conn.commit()
+    return True

backend/tests/conftest.py

@@ -0,0 +1,32 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+import os
+import tempfile
+
+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
+
+
+@pytest.fixture(scope='session', autouse=True)
+def test_users():
+    from backend.app.database import get_connection, hash_password, init_db
+
+    init_db()
+    with get_connection() as conn:
+        conn.executemany(
+            '''INSERT INTO users
+               (id, username, email, password_hash, is_admin, email_verified)
+               VALUES (?, ?, ?, ?, ?, 1)''',
+            [
+                ('user-1', 'alice', 'alice@example.com', hash_password('secret123'), 1),
+                ('user-2', 'bob', 'bob@example.com', hash_password('secret123'), 0),
+            ],
+        )
+        conn.commit()
+    yield
+    TEST_DATABASE_DIRECTORY.cleanup()

backend/tests/test_api.py

@@ -5,10 +5,13 @@ import json
 import threading
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from uuid import uuid4
+from unittest.mock import patch
 
 from fastapi.testclient import TestClient
 
 from backend.app.main import app
+from backend.app.database import get_connection
+from backend.app.services.password_reset import create_reset_token
 from backend.app.services.token_service import issue_token
 
 
@@ -110,6 +113,49 @@ def test_email_verification_link_enables_login():
     assert client.get('/api/auth/verify-email', params={'token': verification_token}).status_code == 400
 
 
+def test_mistyped_password_sends_reset_link_without_changing_login_error():
+    username = f'mistyped-{uuid4().hex}'
+    admin_headers = {'Authorization': f"Bearer {issue_token('user-1', 'alice')['access_token']}"}
+    created = client.post('/api/admin/users', headers=admin_headers, json={
+        'username': username,
+        'email': f'{username}@example.com',
+        'password': 'secret123',
+    })
+    user_id = created.json()['id']
+    with get_connection() as conn:
+        conn.execute('UPDATE users SET email_verified = 1 WHERE id = ?', (user_id,))
+        conn.commit()
+    with patch('backend.app.api.auth.smtp_configured', return_value=True), \
+            patch('backend.app.api.auth.create_reset_token', return_value='reset-token') as create_token, \
+            patch('backend.app.api.auth.send_password_reset_email') as send_email:
+        response = client.post('/api/auth/login', json={'username': username, 'password': 'wrong-password'})
+
+    assert response.status_code == 401
+    assert response.json()['detail'] == 'Invalid username or password'
+    create_token.assert_called_once_with(user_id)
+    send_email.assert_called_once()
+    assert send_email.call_args.args[2].endswith('/reset-password?token=reset-token')
+
+
+def test_password_reset_is_single_use_and_revokes_sessions():
+    username = f'reset-owner-{uuid4().hex}'
+    admin_headers = {'Authorization': f"Bearer {issue_token('user-1', 'alice')['access_token']}"}
+    created = client.post('/api/admin/users', headers=admin_headers, json={
+        'username': username,
+        'email': f'{username}@example.com',
+        'password': 'secret123',
+    })
+    user_id = created.json()['id']
+    with get_connection() as conn:
+        conn.execute('UPDATE users SET email_verified = 1 WHERE id = ?', (user_id,))
+        conn.commit()
+    token = create_reset_token(user_id)
+    reset = client.post('/api/auth/reset-password', json={'token': token, 'password': 'new-secret123'})
+    assert reset.status_code == 200
+    assert client.post('/api/auth/reset-password', json={'token': token, 'password': 'another-secret'}).status_code == 400
+    assert client.post('/api/auth/login', json={'username': username, 'password': 'new-secret123'}).status_code == 200
+
+
 def test_admin_can_remove_user_with_owned_data():
     headers = login_headers()
     username = f'data-owner-{uuid4().hex}'

backend/tests/test_database.py

@@ -10,7 +10,7 @@ def test_database_migrations_are_versioned_and_idempotent():
     connection = sqlite3.connect(':memory:')
 
     apply_migrations(connection)
-    assert get_schema_version(connection) == 6
+    assert get_schema_version(connection) == 7
     tables = {
         row[0]
         for row in connection.execute(
@@ -27,6 +27,6 @@ def test_database_migrations_are_versioned_and_idempotent():
     assert set(DEFAULT_TAGS) <= seeded_tags
 
     apply_migrations(connection)
-    assert get_schema_version(connection) == 6
+    assert get_schema_version(connection) == 7
 
     connection.close()