120 lines
5.2 KiB
Python
120 lines
5.2 KiB
Python
## 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.error import HTTPError
|
|
from urllib.request import Request
|
|
from uuid import uuid4
|
|
|
|
from backend.app.core.config import settings
|
|
from backend.app.database import get_connection
|
|
from backend.app.services.secret_store import decrypt_secret, encrypt_secret
|
|
from backend.app.services.mastodon_security import open_no_redirect, validate_public_instance
|
|
|
|
|
|
def normalize_instance(instance: str) -> str:
|
|
return validate_public_instance(instance)
|
|
|
|
|
|
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 open_no_redirect(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'
|
|
setting_name = f'mastodon_app:{instance}'
|
|
with get_connection() as conn:
|
|
row = conn.execute('SELECT value FROM app_settings WHERE name = ?', (setting_name,)).fetchone()
|
|
app = json.loads(row['value']) if row else None
|
|
if app and app.get('client_secret'):
|
|
app['client_secret'] = decrypt_secret(app['client_secret'])
|
|
if not app:
|
|
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,
|
|
})
|
|
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''',
|
|
(setting_name, json.dumps({**app, 'client_secret': encrypt_secret(app['client_secret'])})),
|
|
)
|
|
conn.commit()
|
|
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'], encrypt_secret(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': decrypt_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': encrypt_secret(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'] |