sevenhill
b4c27da560
Tests / test (push) Canceled after 0s
Native crypto / native-crypto (push) Canceled after 0s
Release (main) / android (oneme) (push) Canceled after 0s
Release (main) / android (qlyra) (push) Canceled after 0s
Release (main) / windows (push) Canceled after 0s
Release (main) / linux (push) Canceled after 0s
Release (main) / macos (push) Canceled after 0s
Release (main) / ios (push) Canceled after 0s
Release (main) / release (push) Canceled after 0s
50 lines
3.1 KiB
Python
50 lines
3.1 KiB
Python
"""Authenticate privately on the Pi and verify the deployed administration API."""
|
|
import json, sqlite3
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
from admin_http import AdminHttp
|
|
|
|
root = Path('/mnt/data/ikar_server')
|
|
settings = dict(line.split('=', 1) for line in (root/'shared/server.env').read_text().splitlines() if '=' in line and not line.startswith('#'))
|
|
base = 'http://127.0.0.1:5099'
|
|
for route in ('users/list', 'reports', 'bots', 'audit', 'operations', 'admin-sessions'):
|
|
try:
|
|
urlopen(base + '/admin/api/' + route, timeout=15)
|
|
raise RuntimeError('Unauthenticated admin access is possible')
|
|
except HTTPError as error:
|
|
if error.code != 401: raise
|
|
client = AdminHttp(base)
|
|
client.login(settings)
|
|
try:
|
|
endpoints = {}
|
|
for route in ('status', 'users/list', 'chats/list', 'attachments/list', 'reports', 'bots', 'audit', 'admin-sessions', 'operations', 'releases', 'push-queue', 'app-updates/compatibility-report'):
|
|
code, data = client.request('/admin/api/' + route)
|
|
if code != 200: raise RuntimeError('Admin endpoint failed: ' + route)
|
|
endpoints[route] = code
|
|
if route == 'admin-sessions' and not any(item['current'] for item in data): raise RuntimeError('Current admin session is missing')
|
|
if route == 'operations':
|
|
if not data['dataProtectionPersistent'] or not data['backups']: raise RuntimeError('Persistent keys or backups are missing')
|
|
operations = {key: data[key] for key in ('diskFreeBytes', 'diskTotalBytes', 'openReports', 'queuedPush')}
|
|
if route == 'releases': release = data['release']
|
|
# A missing CSRF token must fail without terminating the valid session.
|
|
cookies = '; '.join(key + '=' + value for key, value in client.cookies.items())
|
|
try:
|
|
urlopen(Request(base + '/admin/api/session/logout', method='POST', headers={'Cookie': cookies, 'X-Forwarded-Proto': 'https'}), timeout=15)
|
|
raise RuntimeError('A mutation without CSRF was accepted')
|
|
except HTTPError as error:
|
|
if error.code != 400: raise
|
|
if not client.request('/admin/api/session')[1]['authenticated']: raise RuntimeError('CSRF check damaged valid session')
|
|
finally:
|
|
client.logout()
|
|
with sqlite3.connect(f'file:{root}/shared/data/ikar.db?mode=ro', uri=True) as db:
|
|
if db.execute('PRAGMA integrity_check').fetchone()[0] != 'ok': raise RuntimeError('Database integrity failed')
|
|
source = db.execute("SELECT COUNT(*) FROM ModerationLogEntries WHERE Action='complaint.submitted'").fetchone()[0]
|
|
cases = db.execute('SELECT COUNT(*) FROM ModerationCases').fetchone()[0]
|
|
if cases < source: raise RuntimeError('Complaint migration is incomplete')
|
|
report = {'passed': True, 'authenticatedEndpoints': endpoints, 'anonymousDenied': True, 'csrfRequired': True, 'logout': True,
|
|
'databaseIntegrity': 'ok', 'complaintsMigrated': cases, 'operations': operations,
|
|
'androidRelease': {'versionName': release['versionName'], 'versionCode': release['versionCode'], 'sha256': release['sha256']}}
|
|
(root/'admin-console-verification.json').write_text(json.dumps(report, indent=2))
|
|
print(json.dumps(report))
|