first setup configuration
Build LinkLog Development Image / development-image (push) Successful in 10s
Build LinkLog Development Image / development-image (push) Successful in 10s
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
## Copyright © 2026 Olaf Kolkman
|
||||
## SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
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 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
|
||||
|
||||
|
||||
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):
|
||||
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')
|
||||
|
||||
user_id = str(uuid4())
|
||||
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,
|
||||
}
|
||||
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)),
|
||||
)
|
||||
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.'}
|
||||
|
||||
|
||||
@router.get('/status')
|
||||
def setup_status():
|
||||
return {'configured': has_administrator()}
|
||||
+7
-17
@@ -116,6 +116,13 @@ CREATE TABLE IF NOT EXISTS email_verification_tokens (
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_email_verification_tokens_user_id
|
||||
ON email_verification_tokens(user_id);
|
||||
'''),
|
||||
(6, '''
|
||||
CREATE TABLE IF NOT EXISTS app_settings (
|
||||
name TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
'''),
|
||||
]
|
||||
|
||||
@@ -152,23 +159,6 @@ def seed_default_tags(conn: sqlite3.Connection) -> None:
|
||||
def init_db() -> None:
|
||||
with get_connection() as conn:
|
||||
apply_migrations(conn)
|
||||
alice_hash = hash_password('secret123')
|
||||
bob_hash = hash_password('secret123')
|
||||
conn.execute(
|
||||
'''
|
||||
INSERT OR IGNORE INTO users (id, username, email, password_hash, is_admin)
|
||||
VALUES (?, ?, ?, ?, 1)
|
||||
''',
|
||||
('user-1', 'alice', 'alice@example.com', alice_hash)
|
||||
)
|
||||
conn.execute("UPDATE users SET is_admin = 1 WHERE id = 'user-1'")
|
||||
conn.execute(
|
||||
'''
|
||||
INSERT OR IGNORE INTO users (id, username, email, password_hash, is_admin)
|
||||
VALUES (?, ?, ?, ?, 0)
|
||||
''',
|
||||
('user-2', 'bob', 'bob@example.com', bob_hash)
|
||||
)
|
||||
conn.execute(
|
||||
'''
|
||||
INSERT OR IGNORE INTO plugins (id, name, version, enabled, config)
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import HTMLResponse
|
||||
from fastapi.responses import RedirectResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from starlette.requests import Request
|
||||
@@ -11,6 +12,8 @@ from backend.app.api.admin import router as admin_router
|
||||
from backend.app.api.auth import router as auth_router
|
||||
from backend.app.api.links import router as links_router
|
||||
from backend.app.api.public import router as public_router
|
||||
from backend.app.api.setup import router as setup_router
|
||||
from backend.app.api.setup import has_administrator
|
||||
from backend.app.api.user_config import router as user_config_router
|
||||
from backend.app.core.config import settings
|
||||
from backend.app.database import AVATARS_DIR
|
||||
@@ -24,28 +27,37 @@ app.include_router(links_router, prefix='/api')
|
||||
app.include_router(public_router, prefix='/api/public')
|
||||
app.include_router(admin_router, prefix='/api/admin')
|
||||
app.include_router(user_config_router, prefix='/api/user')
|
||||
app.include_router(setup_router, prefix='/api/setup')
|
||||
|
||||
templates = Jinja2Templates(directory='frontend/templates')
|
||||
|
||||
|
||||
@app.get('/', response_class=HTMLResponse)
|
||||
async def public_root(request: Request):
|
||||
if not has_administrator():
|
||||
return RedirectResponse('/setup')
|
||||
feed = list_public_links()
|
||||
return templates.TemplateResponse(request, 'feed.html', {'feed': feed})
|
||||
|
||||
|
||||
@app.get('/admin', response_class=HTMLResponse)
|
||||
async def admin_dashboard(request: Request):
|
||||
if not has_administrator():
|
||||
return RedirectResponse('/setup')
|
||||
return templates.TemplateResponse(request, 'admin.html', {})
|
||||
|
||||
|
||||
@app.get('/profile', response_class=HTMLResponse)
|
||||
async def user_profile_page(request: Request):
|
||||
if not has_administrator():
|
||||
return RedirectResponse('/setup')
|
||||
return templates.TemplateResponse(request, 'user_profile.html', {})
|
||||
|
||||
|
||||
@app.get('/labels', response_class=HTMLResponse)
|
||||
async def labels_page(request: Request):
|
||||
if not has_administrator():
|
||||
return RedirectResponse('/setup')
|
||||
return templates.TemplateResponse(request, 'labels.html', {})
|
||||
|
||||
|
||||
@@ -56,9 +68,18 @@ async def about_page(request: Request):
|
||||
|
||||
@app.get('/login', response_class=HTMLResponse)
|
||||
async def login_page(request: Request):
|
||||
if not has_administrator():
|
||||
return RedirectResponse('/setup')
|
||||
return templates.TemplateResponse(request, 'login.html', {})
|
||||
|
||||
|
||||
@app.get('/setup', response_class=HTMLResponse)
|
||||
async def setup_page(request: Request):
|
||||
if has_administrator():
|
||||
return RedirectResponse('/')
|
||||
return templates.TemplateResponse(request, 'setup.html', {})
|
||||
|
||||
|
||||
@app.get('/health')
|
||||
def health_check():
|
||||
return {'status': 'ok'}
|
||||
|
||||
@@ -3,31 +3,71 @@
|
||||
|
||||
from email.message import EmailMessage
|
||||
from smtplib import SMTP
|
||||
import json
|
||||
|
||||
from backend.app.core.config import settings
|
||||
from backend.app.database import get_connection
|
||||
|
||||
|
||||
def get_smtp_settings() -> dict:
|
||||
values = {
|
||||
'smtp_host': settings.smtp_host,
|
||||
'smtp_port': settings.smtp_port,
|
||||
'smtp_username': settings.smtp_username,
|
||||
'smtp_password': settings.smtp_password,
|
||||
'smtp_from': settings.smtp_from,
|
||||
'smtp_use_tls': settings.smtp_use_tls,
|
||||
}
|
||||
with get_connection() as conn:
|
||||
row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('smtp',)).fetchone()
|
||||
if row:
|
||||
values.update(json.loads(row['value']))
|
||||
return values
|
||||
|
||||
|
||||
def save_smtp_settings(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''',
|
||||
('smtp', json.dumps(values)),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def smtp_configured() -> bool:
|
||||
return bool(settings.smtp_host and settings.smtp_from)
|
||||
smtp = get_smtp_settings()
|
||||
return bool(smtp['smtp_host'] and smtp['smtp_from'])
|
||||
|
||||
|
||||
def send_verification_email(email: str, username: str, verification_url: str) -> None:
|
||||
def send_message(email: str, subject: str, body: str) -> None:
|
||||
smtp = get_smtp_settings()
|
||||
if not smtp_configured():
|
||||
raise RuntimeError('SMTP is not configured; set LINKLOG_SMTP_HOST and LINKLOG_SMTP_FROM')
|
||||
|
||||
message = EmailMessage()
|
||||
message['Subject'] = 'Verify your LinkLog email address'
|
||||
message['From'] = settings.smtp_from
|
||||
message['Subject'] = subject
|
||||
message['From'] = smtp['smtp_from']
|
||||
message['To'] = email
|
||||
message.set_content(
|
||||
message.set_content(body)
|
||||
|
||||
with SMTP(smtp['smtp_host'], smtp['smtp_port'], timeout=10) as connection:
|
||||
if smtp['smtp_use_tls']:
|
||||
connection.starttls()
|
||||
if smtp['smtp_username']:
|
||||
connection.login(smtp['smtp_username'], smtp['smtp_password'])
|
||||
connection.send_message(message)
|
||||
|
||||
|
||||
def send_verification_email(email: str, username: str, verification_url: str) -> None:
|
||||
send_message(
|
||||
email,
|
||||
'Verify your LinkLog email address',
|
||||
f'Hello {username},\n\n'
|
||||
f'Verify your LinkLog email address by opening this link:\n{verification_url}\n\n'
|
||||
f'This link expires in {settings.email_verification_expiry_hours} hours.\n'
|
||||
f'This link expires in {settings.email_verification_expiry_hours} hours.\n',
|
||||
)
|
||||
|
||||
with SMTP(settings.smtp_host, settings.smtp_port, timeout=10) as smtp:
|
||||
if settings.smtp_use_tls:
|
||||
smtp.starttls()
|
||||
if settings.smtp_username:
|
||||
smtp.login(settings.smtp_username, settings.smtp_password)
|
||||
smtp.send_message(message)
|
||||
|
||||
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')
|
||||
Reference in New Issue
Block a user