191 lines
8.0 KiB
Python
191 lines
8.0 KiB
Python
## 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
|
|
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 get_smtp_settings, save_smtp_settings, send_test_email
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class SetupRequest(BaseModel):
|
|
username: str
|
|
email: str
|
|
password: str
|
|
smtp_host: str
|
|
smtp_port: int = 587
|
|
smtp_username: str = ''
|
|
smtp_password: str = ''
|
|
smtp_from: str
|
|
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
|
|
|
|
|
|
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()
|
|
email = payload.email.strip()
|
|
smtp_host = payload.smtp_host.strip()
|
|
smtp_from = payload.smtp_from.strip()
|
|
if not username or not email or not smtp_host or not smtp_from or len(payload.password) < 8:
|
|
raise HTTPException(status_code=422, detail='Admin credentials and SMTP settings are required')
|
|
if not 1 <= payload.smtp_port <= 65535:
|
|
raise HTTPException(status_code=422, detail='SMTP port must be between 1 and 65535')
|
|
|
|
smtp_values = {
|
|
'smtp_host': smtp_host,
|
|
'smtp_port': payload.smtp_port,
|
|
'smtp_username': payload.smtp_username.strip(),
|
|
'smtp_password': payload.smtp_password,
|
|
'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, 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
|
|
delete_pending_setup()
|
|
return {'status': 'configured', 'message': 'LinkLog is configured.'}
|
|
|
|
|
|
@router.get('/status')
|
|
def setup_status():
|
|
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,
|
|
} |