email in admin

backend/app/api/admin.py

@@ -2,6 +2,7 @@
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
 import json
+from datetime import datetime, timedelta, timezone
 from uuid import uuid4
 
 from fastapi import APIRouter, Depends, HTTPException
@@ -10,7 +11,13 @@ from pydantic import BaseModel
 from backend.app.api.dependencies import require_admin
 from backend.app.database import get_connection, hash_password
 from backend.app.services.link_service import delete_label
-from backend.app.services.email_service import send_verification_email, smtp_configured
+from backend.app.services.email_service import (
+    get_smtp_settings,
+    save_smtp_settings,
+    send_test_email,
+    send_verification_email,
+    smtp_configured,
+)
 from backend.app.services.email_verification import create_verification_token
 from backend.app.core.config import settings
 
@@ -33,6 +40,32 @@ class AdminUserUpdate(BaseModel):
     is_admin: bool
 
 
+class AdminSmtpUpdate(BaseModel):
+    smtp_host: str
+    smtp_port: int = 587
+    smtp_username: str = ''
+    smtp_password: str = ''
+    smtp_from: str
+    smtp_use_tls: bool = True
+
+
+def validate_smtp_values(payload: AdminSmtpUpdate, current: dict | None = None) -> dict:
+    smtp_host = payload.smtp_host.strip()
+    smtp_from = payload.smtp_from.strip()
+    if not smtp_host or not smtp_from:
+        raise HTTPException(status_code=422, detail='SMTP host and sender address are required')
+    if not 1 <= payload.smtp_port <= 65535:
+        raise HTTPException(status_code=422, detail='SMTP port must be between 1 and 65535')
+    return {
+        'smtp_host': smtp_host,
+        'smtp_port': payload.smtp_port,
+        'smtp_username': payload.smtp_username.strip(),
+        'smtp_password': payload.smtp_password or (current or {}).get('smtp_password', ''),
+        'smtp_from': smtp_from,
+        'smtp_use_tls': payload.smtp_use_tls,
+    }
+
+
 def public_user(row):
     return {
         'id': row['id'],
@@ -90,6 +123,74 @@ def create_user(payload: AdminUserCreate, _: dict = Depends(require_admin)):
     return public_user(row)
 
 
+def public_smtp_settings(values: dict) -> dict:
+    return {
+        'smtp_host': values['smtp_host'],
+        'smtp_port': values['smtp_port'],
+        'smtp_username': values['smtp_username'],
+        'smtp_from': values['smtp_from'],
+        'smtp_use_tls': values['smtp_use_tls'],
+        'password_configured': bool(values['smtp_password']),
+    }
+
+
+@router.get('/smtp')
+def get_admin_smtp_settings(_: dict = Depends(require_admin)):
+    return public_smtp_settings(get_smtp_settings())
+
+
+@router.put('/smtp')
+def update_admin_smtp_settings(payload: AdminSmtpUpdate, _: dict = Depends(require_admin)):
+    current = get_smtp_settings()
+    values = validate_smtp_values(payload, current)
+    save_smtp_settings(values)
+    return public_smtp_settings(values)
+
+
+@router.post('/smtp/test')
+def validate_admin_smtp(payload: AdminSmtpUpdate, current_user: dict = Depends(require_admin)):
+    values = validate_smtp_values(payload, get_smtp_settings())
+    now = datetime.now(timezone.utc)
+    with get_connection() as conn:
+        row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('admin_smtp_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'SMTP validation 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 validation email.', headers={'Retry-After': str(retry_after)})
+    try:
+        send_test_email(current_user['email'], values)
+    except Exception as error:
+        raise HTTPException(status_code=503, detail=f'SMTP validation failed: {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''',
+            ('admin_smtp_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': f'SMTP validation email sent to {current_user["email"]}.',
+        'sends_remaining': max(0, 5 - sends),
+        'cooldown_seconds': 120 if sends >= 5 else 0,
+        'next_allowed_at': next_allowed.isoformat(),
+    }
+
+
 @router.put('/users/{user_id}')
 def update_user_privileges(
     user_id: str,

backend/app/services/email_service.py

@@ -40,8 +40,8 @@ def smtp_configured() -> bool:
     return bool(smtp['smtp_host'] and smtp['smtp_from'])
 
 
-def send_message(email: str, subject: str, body: str) -> None:
-    smtp = get_smtp_settings()
+def send_message(email: str, subject: str, body: str, smtp_values: dict | None = None) -> None:
+    smtp = smtp_values or get_smtp_settings()
     if not smtp_configured():
         raise RuntimeError('SMTP is not configured; set LINKLOG_SMTP_HOST and LINKLOG_SMTP_FROM')
 
@@ -69,8 +69,13 @@ 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')
+def send_test_email(email: str, smtp_values: dict | None = None) -> None:
+    send_message(
+        email,
+        'LinkLog SMTP test',
+        'This is a test message from LinkLog. SMTP is configured correctly.\n',
+        smtp_values,
+    )
 
 
 def send_password_reset_email(email: str, username: str, reset_url: str) -> None:

backend/tests/test_api.py

@@ -12,6 +12,7 @@ from fastapi.testclient import TestClient
 
 from backend.app.main import app
 from backend.app.database import get_connection
+from backend.app.services.email_service import get_smtp_settings
 from backend.app.services.password_reset import create_reset_token
 from backend.app.services.token_service import issue_token
 
@@ -57,6 +58,64 @@ def test_configuration_requires_authentication_and_admin_role():
     assert client.get('/api/admin/users').status_code == 401
     assert client.get('/api/admin/plugins', headers=login_headers('bob')).status_code == 403
     assert client.get('/api/admin/users', headers=login_headers('bob')).status_code == 403
+    assert client.get('/api/admin/smtp').status_code == 401
+    assert client.get('/api/admin/smtp', headers=login_headers('bob')).status_code == 403
+
+
+def test_admin_can_save_and_validate_smtp_settings():
+    headers = login_headers()
+    original = get_smtp_settings()
+    response = client.put('/api/admin/smtp', headers=headers, json={
+        '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,
+    })
+    assert response.status_code == 200
+    assert response.json()['smtp_host'] == 'smtp.example.com'
+    assert response.json()['password_configured'] is True
+    assert 'smtp_password' not in response.json()
+
+    with patch('backend.app.api.admin.send_test_email') as send_test_email:
+        validation = client.post('/api/admin/smtp/test', headers=headers, json={
+            'smtp_host': 'smtp.unsaved.example.com',
+            'smtp_port': 2525,
+            'smtp_username': 'temporary-user',
+            'smtp_password': 'temporary-secret',
+            'smtp_from': 'Temporary <temporary@example.com>',
+            'smtp_use_tls': False,
+        })
+        assert validation.status_code == 200
+        send_test_email.assert_called_once_with('alice@example.com', {
+            'smtp_host': 'smtp.unsaved.example.com',
+            'smtp_port': 2525,
+            'smtp_username': 'temporary-user',
+            'smtp_password': 'temporary-secret',
+            'smtp_from': 'Temporary <temporary@example.com>',
+            'smtp_use_tls': False,
+        })
+
+    client.put('/api/admin/smtp', headers=headers, json=original)
+
+
+def test_admin_reports_smtp_validation_errors():
+    headers = login_headers()
+    with get_connection() as conn:
+        conn.execute('DELETE FROM app_settings WHERE name = ?', ('admin_smtp_mail_rate',))
+        conn.commit()
+    with patch('backend.app.api.admin.send_test_email', side_effect=RuntimeError('connection refused')):
+        failed_validation = client.post('/api/admin/smtp/test', headers=headers, json={
+            'smtp_host': 'smtp.unsaved.example.com',
+            'smtp_port': 2525,
+            'smtp_username': 'temporary-user',
+            'smtp_password': 'temporary-secret',
+            'smtp_from': 'Temporary <temporary@example.com>',
+            'smtp_use_tls': False,
+        })
+    assert failed_validation.status_code == 503
+    assert 'connection refused' in failed_validation.json()['detail']
 
 
 def test_admin_can_add_list_and_remove_users():
@@ -66,6 +125,7 @@ def test_admin_can_add_list_and_remove_users():
         'email': 'charlie@example.com',
         'password': 'charlie-secret',
     })
+
     assert create_response.status_code == 201
     user = create_response.json()
     assert user['username'] == 'charlie'