TPSchedule/server/updates/tests/test_updates.py

import re
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from update_service import Settings, create_app, hash_password


BASE = 'https://app.glowtps.cn'
PASSWORD = 'only-for-tests-password'


@pytest.fixture
def service(tmp_path):
    settings = Settings(data_dir=str(tmp_path), public_origin=BASE,
                        admin_username='admin', admin_password_hash=hash_password(PASSWORD),
                        session_secret='test-session-secret-at-least-32-characters')
    with TestClient(create_app(settings), base_url=BASE) as client:
        yield client, settings


def csrf(response):
    return re.search(r'name="csrf" value="([a-f0-9]+)"', response.text)[1]


def login(client):
    page = client.get('/update')
    return client.post('/update/login', data={'csrf': csrf(page), 'username': 'admin', 'password': PASSWORD},
                       headers={'Origin': BASE}, follow_redirects=False)


def publish(client, **changes):
    page = client.get('/update')
    fields = {'csrf': csrf(page), 'version': '0.8.0', 'build_number': '25',
              'notes': '新增自动检查更新\n修复课表导入', 'download_url': BASE + '/downloads/TPSchedule.apk',
              'revision': re.search(r'name="revision" value="([a-f0-9]+|none)"', page.text)[1]}
    fields.update(changes)
    return client.post('/update/publish', data=fields, headers={'Origin': BASE}, follow_redirects=False)


def test_private_admin_and_initial_public_empty_manifest(service):
    client, _ = service
    page = client.get('/update')
    assert '登录更新后台' in page.text
    assert 'name="download_url"' not in page.text
    assert 'HttpOnly' in page.headers['set-cookie'] and 'Secure' in page.headers['set-cookie']
    assert client.get('/v1/updates/android').json() == {'schema_version': 1, 'application_id': 'com.tpschedule.app', 'latest': None}
    assert client.post('/update/publish', data={}).status_code == 401


def test_publish_persists_and_app_reads_exact_public_fields(service):
    client, settings = service
    assert login(client).status_code == 303
    assert publish(client).status_code == 303
    result = client.get('/v1/updates/android')
    release = result.json()['latest']
    assert release['version'] == '0.8.0' and release['build_number'] == 25
    assert release['notes'] == '新增自动检查更新\n修复课表导入'
    assert release['download_url'] == BASE + '/downloads/TPSchedule.apk'
    assert 'password' not in result.text and 'session_secret' not in result.text
    assert result.headers['cache-control'] == 'no-store'
    with TestClient(create_app(settings), base_url=BASE) as restarted:
        assert restarted.get('/v1/updates/android').json() == result.json()


@pytest.mark.parametrize('changes', [
    {'download_url': 'javascript:alert(1)'}, {'download_url': 'http://example.com/a.apk'},
    {'download_url': 'https://user:password@example.com/a.apk'},
    {'notes': ''}, {'version': 'V1.0'}, {'build_number': '0'}, {'build_number': '2147483648'},
    {'notes': 'x' * 8001},
    {'version': '１.０.０'},
])
def test_bad_publish_does_not_replace_release(service, changes):
    client, _ = service
    login(client)
    publish(client)
    previous = client.get('/v1/updates/android').json()
    assert publish(client, **changes).status_code == 422
    assert client.get('/v1/updates/android').json() == previous


def test_validation_failure_retains_typed_release_notes(service):
    client, _ = service
    login(client)
    response = publish(client, notes='尚未发布的完整说明', download_url='http://bad.test/app.apk')
    assert response.status_code == 422
    assert '尚未发布的完整说明' in response.text
    assert 'http://bad.test/app.apk' in response.text


def test_csrf_and_cross_origin_login_publish_logout_are_rejected(service):
    client, _ = service
    page = client.get('/update')
    assert client.post('/update/login', data={'username': 'admin', 'password': PASSWORD, 'csrf': csrf(page)},
                       headers={'Origin': 'https://evil.test'}).status_code == 403
    assert login(client).status_code == 303
    assert publish(client, csrf='bad').status_code == 403
    assert publish(client, csrf='无效').status_code == 403
    assert publish(client, revision='无效').status_code == 422
    assert client.post('/update/logout', data={'csrf': 'bad'}, headers={'Origin': BASE}).status_code == 403


@pytest.mark.parametrize('origin', [None, 'null', 'https://evil.test'])
def test_valid_csrf_does_not_allow_missing_null_or_foreign_origin(service, origin):
    client, _ = service
    headers = {} if origin is None else {'Origin': origin}
    page = client.get('/update')
    values = {'username': 'admin', 'password': PASSWORD, 'csrf': csrf(page)}
    assert client.post('/update/login', data=values, headers=headers).status_code == 403
    assert login(client).status_code == 303
    page = client.get('/update')
    values = {'csrf': csrf(page), 'version': '0.8.0', 'build_number': '25',
              'notes': '不得发布', 'download_url': BASE + '/test.apk', 'revision': 'none'}
    assert client.post('/update/publish', data=values, headers=headers).status_code == 403
    assert client.get('/v1/updates/android').json()['latest'] is None
    assert client.post('/update/logout', data={'csrf': csrf(page)}, headers=headers).status_code == 403
    assert 'name="download_url"' in client.get('/update').text


def test_login_rejects_unicode_username_and_limits_password_guesses(service):
    client, _ = service
    page = client.get('/update')
    response = client.post('/update/login', data={'csrf': csrf(page), 'username': '管理员', 'password': PASSWORD}, headers={'Origin': BASE})
    assert response.status_code == 401
    for _ in range(9):
        page = client.get('/update')
        response = client.post('/update/login', data={'csrf': csrf(page), 'username': 'admin', 'password': 'wrong'}, headers={'Origin': BASE})
        assert response.status_code == 401
    assert login(client).status_code == 429


def test_tampered_cookie_cannot_publish_and_oversized_body_is_rejected(service):
    client, _ = service
    login(client)
    cookie = client.cookies.get('tpschedule_update_session')
    client.cookies.clear()
    client.cookies.set('tpschedule_update_session', cookie + 'tampered', path='/update')
    assert client.post('/update/publish', data={}).status_code == 401
    assert client.post('/update/login', content=b'x' * (96 * 1024 + 1)).status_code == 413


def test_monotonic_build_and_stale_form_do_not_overwrite_published_version(service):
    client, _ = service
    login(client)
    assert publish(client).status_code == 303
    assert publish(client, version='0.7.9', build_number='24').status_code == 422
    assert publish(client, version='0.8.1', build_number='25').status_code == 422
    assert publish(client, version='0.8.1', build_number='26', revision='none').status_code == 409
    assert publish(client, version='0.8.1', build_number='26').status_code == 303


def test_release_notes_are_escaped_in_admin_and_login_does_not_echo_password(service):
    client, _ = service
    page = client.get('/update')
    bad = client.post('/update/login', data={'csrf': csrf(page), 'username': 'admin', 'password': 'wrong-unique-password'}, headers={'Origin': BASE})
    assert bad.status_code == 401 and 'wrong-unique-password' not in bad.text
    login(client)
    publish(client, notes='<script>alert(1)</script>')
    page = client.get('/update')
    assert '<script>alert(1)</script>' not in page.text
    assert '&lt;script&gt;' in page.text


def test_logout_invalidates_browser_session_and_corrupt_file_is_not_no_update(service):
    client, settings = service
    login(client)
    page = client.get('/update')
    assert client.post('/update/logout', data={'csrf': csrf(page)}, headers={'Origin': BASE}, follow_redirects=False).status_code == 303
    assert '登录更新后台' in client.get('/update').text
    Path(settings.data_dir, 'android-release.json').write_text('broken', encoding='utf-8')
    assert client.get('/v1/updates/android').status_code == 503