111 lines
5.7 KiB
Python
111 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Launch the built binary against disposable synthetic data; never use a real DB."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
def main():
|
|
root = Path(__file__).resolve().parents[1]
|
|
binary = Path(sys.argv[1] if len(sys.argv) > 1 else root / 'target/debug/twofauth-rs').resolve()
|
|
with tempfile.TemporaryDirectory(prefix='twofauth-rs-smoke-') as folder:
|
|
work = Path(folder)
|
|
password = work / 'password'
|
|
password.write_text('Synthetic-password-4620')
|
|
password.chmod(0o600)
|
|
env = dict(os.environ, DATABASE_URL=f'sqlite://{work}/test.sqlite', RUST_COOKIE_SECURE='false', RUST_PUBLIC_DIR=str(root / 'public'))
|
|
env.pop('APP_SUBDIRECTORY', None)
|
|
env.pop('APP_KEY', None)
|
|
env.pop('APP_PREVIOUS_KEYS', None)
|
|
subprocess.run([str(binary), 'init'], env=env, check=True, capture_output=True, cwd=work)
|
|
duplicate = subprocess.run([str(binary), 'init'], env=env, capture_output=True, cwd=work)
|
|
assert duplicate.returncode != 0, 'init must refuse an existing users table'
|
|
subprocess.run([str(binary), 'create-user', '--name', 'Synthetic', '--email', 'smoke@example.invalid', '--password-file', str(password)], env=env, check=True, capture_output=True, cwd=work)
|
|
password.unlink()
|
|
with socket.socket() as reserve:
|
|
reserve.bind(('127.0.0.1', 0))
|
|
port = reserve.getsockname()[1]
|
|
base = f'http://127.0.0.1:{port}'
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
process = subprocess.Popen([str(binary), 'serve', '--listen', f'127.0.0.1:{port}'], env=env, cwd=work, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
def request(path, method='GET', value=None, csrf=None):
|
|
data = None if value is None else json.dumps(value).encode()
|
|
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
|
|
if csrf is not None:
|
|
headers['X-XSRF-TOKEN'] = csrf
|
|
req = urllib.request.Request(base + path, data=data, headers=headers, method=method)
|
|
try:
|
|
response = opener.open(req, timeout=10)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
with response:
|
|
body = response.read()
|
|
try:
|
|
payload = json.loads(body)
|
|
except (ValueError, UnicodeDecodeError):
|
|
payload = body.decode()
|
|
return response.status, payload, response.headers
|
|
for _ in range(100):
|
|
if process.poll() is not None:
|
|
raise AssertionError('server exited during startup')
|
|
try:
|
|
if request('/up')[0] == 200:
|
|
break
|
|
except urllib.error.URLError:
|
|
time.sleep(0.05)
|
|
else:
|
|
raise AssertionError('server startup timed out')
|
|
assert request('/api/v1/twofaccounts')[0] == 401
|
|
status, html, headers = request('/')
|
|
assert status == 200 and 'var appSettings' in html
|
|
assert headers['Content-Security-Policy']
|
|
csrf = next(cookie.value for cookie in jar if cookie.name == 'XSRF-TOKEN')
|
|
assert request('/user/login', 'POST', {'email': 'smoke@example.invalid', 'password': 'Synthetic-password-4620'})[0] == 419
|
|
status, login, _ = request('/user/login', 'POST', {'email': 'SMOKE@example.invalid', 'password': 'Synthetic-password-4620'}, csrf)
|
|
assert status == 200
|
|
csrf = next(cookie.value for cookie in jar if cookie.name == 'XSRF-TOKEN')
|
|
assert request('/api/v1/user')[1]['email'] == 'smoke@example.invalid'
|
|
account = {'otp_type': 'hotp', 'account': 'Synthetic', 'secret': base64.b32encode(b'12345678901234567890').decode(), 'counter': 0, 'digits': 6, 'algorithm': 'sha1'}
|
|
assert request('/api/v1/twofaccounts', 'POST', account)[0] == 419
|
|
status, created, _ = request('/api/v1/twofaccounts', 'POST', account, csrf)
|
|
assert status == 201
|
|
account_id = created['id']
|
|
listed = request('/api/v1/twofaccounts')[1]
|
|
assert isinstance(listed, list) and len(listed) == 1 and 'secret' not in listed[0]
|
|
for counter in range(2):
|
|
status, otp, _ = request(f'/api/v1/twofaccounts/{account_id}/otp')
|
|
digest = hmac.new(b'12345678901234567890', counter.to_bytes(8, 'big'), hashlib.sha1).digest()
|
|
offset = digest[-1] & 15
|
|
expected = f'{(int.from_bytes(digest[offset:offset+4], "big") & 0x7fffffff) % 1000000:06d}'
|
|
assert status == 200 and otp['password'] == expected
|
|
assert request('/api/v1/settings')[0] == 403
|
|
assert request('/user/logout')[0] == 419
|
|
assert request('/user/logout', csrf=csrf)[0] == 200
|
|
assert request('/api/v1/twofaccounts')[0] == 401
|
|
print(json.dumps({'cli_init': 'passed', 'existing_db_refusal': 'passed', 'test_user': 'passed', 'http_login_crud_hotp_logout': 'passed', 'php_differential': 'not_run'}))
|
|
finally:
|
|
process.send_signal(signal.SIGINT)
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|