Mastodon Oauth

backend/app/api/mastodon.py

@@ -0,0 +1,33 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+from urllib.parse import quote
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import RedirectResponse
+from starlette.requests import Request
+
+from backend.app.api.dependencies import get_current_user
+from backend.app.services.mastodon_oauth import finish_authorization, start_authorization
+
+router = APIRouter()
+
+
+@router.get('/oauth/start')
+def oauth_start(instance: str = 'mastodon.social', user: dict = Depends(get_current_user)):
+    try:
+        authorization_url = start_authorization(user['id'], instance)
+    except Exception as error:
+        raise HTTPException(status_code=502, detail=f'Could not register with Mastodon: {error}') from error
+    return {'authorization_url': authorization_url}
+
+
+@router.get('/oauth/callback')
+def oauth_callback(request: Request, code: str | None = None, state: str | None = None, error: str | None = None):
+    if error or not code or not state:
+        return RedirectResponse(f'/profile?mastodon_error={quote(error or "Authorization was cancelled")}')
+    try:
+        finish_authorization(code, state)
+    except Exception as callback_error:
+        return RedirectResponse(f'/profile?mastodon_error={quote(str(callback_error))}')
+    return RedirectResponse('/profile?mastodon=connected')

backend/app/core/config.py

@@ -26,6 +26,8 @@ class Settings:
     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'))
+    mastodon_client_name: str = os.getenv('LINKLOG_MASTODON_CLIENT_NAME', 'LinkLog')
+    mastodon_oauth_expiry_minutes: int = int(os.getenv('LINKLOG_MASTODON_OAUTH_EXPIRY_MINUTES', '10'))
     tracking_params: list[str] = None
 
     def __post_init__(self):

backend/app/database.py

@@ -135,6 +135,22 @@ CREATE TABLE IF NOT EXISTS password_reset_tokens (
 );
 CREATE INDEX IF NOT EXISTS idx_password_reset_tokens_user_id
     ON password_reset_tokens(user_id);
+'''),
+    (8, '''
+CREATE TABLE IF NOT EXISTS mastodon_oauth_states (
+    id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    state_hash TEXT NOT NULL UNIQUE,
+    instance TEXT NOT NULL,
+    client_id TEXT NOT NULL,
+    client_secret TEXT NOT NULL,
+    redirect_uri TEXT NOT NULL,
+    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_mastodon_oauth_states_state_hash
+    ON mastodon_oauth_states(state_hash);
 '''),
 ]
 

backend/app/main.py

@@ -11,6 +11,7 @@ from starlette.requests import Request
 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.mastodon import router as mastodon_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
@@ -24,6 +25,7 @@ app.mount('/static', StaticFiles(directory='frontend/static'), name='static')
 app.mount('/media', StaticFiles(directory=AVATARS_DIR), name='media')
 app.include_router(auth_router, prefix='/api/auth')
 app.include_router(links_router, prefix='/api')
+app.include_router(mastodon_router, prefix='/api/mastodon')
 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')

backend/app/services/mastodon_oauth.py

@@ -0,0 +1,106 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+from datetime import datetime, timedelta, timezone
+from hashlib import sha256
+import json
+from secrets import token_urlsafe
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+from uuid import uuid4
+
+from backend.app.core.config import settings
+from backend.app.database import get_connection
+
+
+def normalize_instance(instance: str) -> str:
+    value = instance.strip().rstrip('/')
+    if not value:
+        raise ValueError('Mastodon instance is required')
+    return value if '://' in value else f'https://{value}'
+
+
+def post_form(url: str, values: dict) -> dict:
+    request = Request(
+        url,
+        data=urlencode(values).encode('utf-8'),
+        headers={'Content-Type': 'application/x-www-form-urlencoded'},
+        method='POST',
+    )
+    with urlopen(request, timeout=10) as response:
+        return json.loads(response.read().decode('utf-8'))
+
+
+def start_authorization(user_id: str, instance: str) -> str:
+    instance = normalize_instance(instance)
+    redirect_uri = f'{settings.public_url}/api/mastodon/oauth/callback'
+    app = post_form(f'{instance}/api/v1/apps', {
+        'client_name': settings.mastodon_client_name,
+        'redirect_uris': redirect_uri,
+        'scopes': 'read:accounts write:statuses',
+        'website': settings.public_url,
+    })
+    state = token_urlsafe(32)
+    expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.mastodon_oauth_expiry_minutes)
+    with get_connection() as conn:
+        conn.execute('DELETE FROM mastodon_oauth_states WHERE user_id = ?', (user_id,))
+        conn.execute(
+            '''INSERT INTO mastodon_oauth_states
+               (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()),
+        )
+        conn.commit()
+    return f'{instance}/oauth/authorize?' + urlencode({
+        'client_id': app['client_id'],
+        'redirect_uri': redirect_uri,
+        'response_type': 'code',
+        'scope': 'read:accounts write:statuses',
+        'state': state,
+    })
+
+
+def finish_authorization(code: str, state: str) -> str:
+    now = datetime.now(timezone.utc).isoformat()
+    with get_connection() as conn:
+        record = conn.execute(
+            '''SELECT * FROM mastodon_oauth_states
+               WHERE state_hash = ? AND expires_at > ?''',
+            (sha256(state.encode()).hexdigest(), now),
+        ).fetchone()
+        if record is None:
+            raise ValueError('OAuth state is invalid or expired')
+        conn.execute('DELETE FROM mastodon_oauth_states WHERE id = ?', (record['id'],))
+        conn.commit()
+    token = post_form(f"{record['instance']}/oauth/token", {
+        'grant_type': 'authorization_code',
+        'code': code,
+        'client_id': record['client_id'],
+        'client_secret': record['client_secret'],
+        'redirect_uri': record['redirect_uri'],
+    })
+    access_token = token.get('access_token')
+    if not access_token:
+        raise ValueError('Mastodon did not return an access token')
+    with get_connection() as conn:
+        current = conn.execute(
+            'SELECT config FROM user_plugin_config WHERE user_id = ? AND plugin_name = ?',
+            (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})
+        if current:
+            conn.execute(
+                'UPDATE user_plugin_config SET config = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? AND plugin_name = ?',
+                (json.dumps(config), record['user_id'], 'mastodon'),
+            )
+        else:
+            conn.execute(
+                '''INSERT INTO user_plugin_config
+                   (id, user_id, plugin_name, config, created_at, updated_at)
+                   VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)''',
+                (str(uuid4()), record['user_id'], 'mastodon', json.dumps(config)),
+            )
+        conn.commit()
+    return record['user_id']

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) == 7
+    assert get_schema_version(connection) == 8
     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) == 7
+    assert get_schema_version(connection) == 8
 
     connection.close()