generated from jric11/baseProject
128 lines
8.2 KiB
Python
128 lines
8.2 KiB
Python
"""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': 'print', 'report': 'Email', '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)
|