huntingclubroster/tests/auth_http.py

"""End-to-end tests against a disposable local database and PHP development server.

Set HCR_TEST_DSN, HCR_TEST_USER, HCR_TEST_PASSWORD, HCR_DB_HOST,
HCR_DB_PORT, HCR_DB_USER, HCR_DB_PASSWORD and PHP_TEST_COMMAND (JSON argv).
"""
import http.cookiejar
import json
import os
from pathlib import Path
import re
import socket
import subprocess
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request

ROOT = Path(__file__).resolve().parents[1]
PHP = json.loads(os.environ.get('PHP_TEST_COMMAND', '["php"]'))
env = os.environ.copy()
env['HCR_APPLICATION_URL'] = ''  # Never send real email from the test suite.
env['HCR_ENVIRONMENT'] = 'Development'
checks = []

def fixture(action, database=None):
    args = PHP + ['tests/http_fixture.php', action] + ([database] if database else [])
    result = subprocess.run(args, cwd=ROOT, env=env, check=True, text=True, capture_output=True)
    return json.loads(result.stdout.splitlines()[-1]) if result.stdout.strip() else {}

def check(condition, message):
    if not condition:
        raise AssertionError(message)
    checks.append(message)

jar = http.cookiejar.CookieJar()
client = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))

def request(path='/', data=None, headers=None):
    encoded = urllib.parse.urlencode(data).encode() if data is not None else None
    req = urllib.request.Request(base + path, data=encoded, headers=headers or {})
    try:
        with client.open(req, timeout=20) as response:
            return response.status, response.read().decode(), response.headers
    except urllib.error.HTTPError as error:
        return error.code, error.read().decode(), error.headers

def csrf(html):
    match = re.search(r'name="csrf_token"[^>]*value="([a-f0-9]+)"', html)
    if not match:
        match = re.search(r'name="csrf-token"[^>]*content="([a-f0-9]+)"', html)
    if not match:
        raise AssertionError('Missing CSRF token')
    return match.group(1)

database = fixture('create')['database']
env['HCR_DB_NAME'] = database
with socket.socket() as available:
    available.bind(('127.0.0.1', 0))
    port = available.getsockname()[1]
base = f'http://127.0.0.1:{port}'
server = None
try:
    result = subprocess.run(PHP + ['bin/migrate-auth.php'], cwd=ROOT, env=env, check=True, capture_output=True, text=True)
    check('schema is ready' in result.stdout, 'Database migration can be rerun safely')
    with tempfile.TemporaryFile(mode='w+b') as log:
        server = subprocess.Popen(PHP + ['-S', f'127.0.0.1:{port}', 'bin/dev-router.php'], cwd=ROOT, env=env, stdout=log, stderr=log)
        for _ in range(50):
            try:
                status, html, headers = request()
                break
            except urllib.error.URLError:
                time.sleep(0.1)
        else:
            raise AssertionError('PHP development server did not start')
        check(status == 200 and 'name="user_email"' in html, 'Login page renders over HTTP')
        check('no-store' in headers.get('Cache-Control', ''), 'Authentication pages are not cached')
        check('HttpOnly' in headers.get('Set-Cookie', '') and 'SameSite=Lax' in headers.get('Set-Cookie', ''), 'Session cookies have HttpOnly and SameSite protections')
        check(request(data={'action': 'SignIn.login', 'user_email': 'http@example.test', 'user_password': 'wrong'})[0] == 403, 'Login rejects missing CSRF token')
        check(request(data={'action[]': 'bad'})[0] == 400, 'Array-shaped request fields are rejected')
        status, html, _ = request('/?Xaction=signup')
        check(status == 200 and 'name="user_first_name"' in html, 'Signup is reachable before login')
        fields = {'action': 'SignUp.createAccount', 'csrf_token': csrf(html), 'user_first_name': 'HTTP', 'user_last_name': 'Tester', 'user_email': 'http@example.test', 'user_name': 'httpuser', 'user_password': 'HTTP long password!', 'user_password_confirm': 'HTTP long password!', 'user_role': 'Administrator'}
        status, html, _ = request(data=fields)
        check(status == 200 and 'Your account has been created' in html, 'Signup form creates an account and redirects')
        status, html, _ = request()
        status, html, _ = request(data={'action': 'SignIn.login', 'csrf_token': csrf(html), 'user_email': 'http@example.test', 'user_password': 'wrong'})
        check(status == 200 and 'Invalid email address or password' in html, 'Invalid login renders the existing login template')
        status, html, _ = request(data={'action': 'SignIn.login', 'csrf_token': csrf(html), 'user_email': 'http@example.test', 'user_password': 'HTTP long password!'})
        check(status == 200 and 'name="csrf-token"' in html, 'Email/password form signs in and reaches the authenticated shell')
        token = csrf(html)
        check(request(data={'action': 'Users.editUser', 'step': 'update', 'csrf_token': token, 'user_serial': '1', 'user_role': 'Administrator'})[0] == 403, 'Regular users cannot edit account privileges')
        check(request(data={'action': 'RosterReports.csv', 'club_serial': '1', 'csrf_token': token})[0] == 403, 'Regular users cannot invoke administrative report actions')
        check(request(data={'action': 'Base.executeSQL', 'csrf_token': token})[0] == 404, 'Arbitrary helper methods cannot be dispatched')
        check(request(data={'action': 'signout'})[0] == 403, 'Logout rejects missing CSRF token')
        status, html, _ = request(data={'action': 'signout', 'csrf_token': token})
        check(status == 200 and 'id="login"' in html, 'Logout ends the authenticated session')
        status, html, _ = request('/?Xaction=resetpassword')
        check('Send reset link' in html, 'Password reset request page is reachable before login')
        status, html, _ = request(data={'action': 'resetpassword', 'csrf_token': csrf(html), 'user_email': 'unknown@example.test'})
        check(status == 200 and 'If an active account matches' in html, 'Unknown account reset returns a neutral response')
        status, form, _ = request('/?Xaction=resetpassword')
        status, html, _ = request(data={'action': 'resetpassword', 'csrf_token': csrf(form), 'user_email': 'http@example.test'})
        check('If an active account matches' in html, 'Reset delivery/configuration failure returns the same neutral response')
        reset_token = fixture('token', database)['token']
        status, html, headers = request('/?Xaction=resetpassword&token=' + reset_token)
        check(status == 200 and 'Set new password' in html and headers.get('Referrer-Policy') == 'no-referrer', 'Reset link renders a protected new-password form')
        status, html, _ = request(data={'action': 'resetpassword', 'step': 'reset', 'token': reset_token, 'csrf_token': csrf(html), 'user_password': 'HTTP changed password!', 'user_password_confirm': 'HTTP changed password!'})
        check(status == 200 and 'Your password has been changed' in html, 'Reset form changes the password and redirects')
        status, html, _ = request('/?Xaction=resetpassword&token=' + reset_token)
        status, html, _ = request(data={'action': 'resetpassword', 'step': 'reset', 'token': reset_token, 'csrf_token': csrf(html), 'user_password': 'HTTP changed password!', 'user_password_confirm': 'HTTP changed password!'})
        check('invalid or expired' in html, 'Used reset link fails through the real HTTP handler')
        status, html, _ = request()
        status, html, _ = request(data={'action': 'SignIn.login', 'csrf_token': csrf(html), 'user_email': 'http@example.test', 'user_password': 'HTTP changed password!'})
        check(status == 200 and 'name="csrf-token"' in html, 'Reset password works through the real login form')
        fixture('revoke', database)
        check(request(data={'action': 'toggleTheme', 'csrf_token': csrf(html)}, headers={'X-Requested-With': 'XMLHttpRequest'})[0] == 401, 'Revoked AJAX sessions return 401 instead of a login document')
        for path in ['/xml/settings.local.xml', '/tables/_createSystemTable.php', '/bin/migrate-auth.php', '/.git/config']:
            check(request(path)[0] == 404, 'Development router blocks ' + path)
        for message in checks:
            print('PASS:', message)
        print(len(checks), 'HTTP checks passed.')
finally:
    if server:
        server.terminate()
        server.wait(timeout=10)
    fixture('drop', database)