generated from jric11/baseProject
84 lines
4.9 KiB
Python
84 lines
4.9 KiB
Python
"""Disposable HTTP test application. Requires an explicitly supplied test database server."""
|
|
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"]'))
|
|
|
|
class Application:
|
|
def __init__(self):
|
|
if not os.environ.get('HCR_TEST_DSN'):
|
|
raise RuntimeError('An isolated HCR_TEST_DSN is required.')
|
|
self.env=os.environ.copy()
|
|
self.env['HCR_MFA_KEY']='AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA='
|
|
self.env.update(HCR_SUPER_ADMIN_EMAIL='',HCR_APPLICATION_URL='',HCR_ENVIRONMENT='Development',HCR_SMTP_HOST='127.0.0.1',HCR_SMTP_PORT='1',HCR_SMTP_USER='test@example.test',HCR_SMTP_PASSWORD='test-only',HCR_SMTP_FROM='test@example.test')
|
|
self.database=self.fixture('create')['database']
|
|
self.env['HCR_DB_NAME']=self.database
|
|
with socket.socket() as available:
|
|
available.bind(('127.0.0.1',0));port=available.getsockname()[1]
|
|
self.base=f'http://127.0.0.1:{port}'
|
|
self.log=tempfile.TemporaryFile(mode='w+b')
|
|
self.server=subprocess.Popen(PHP+['-S',f'127.0.0.1:{port}','bin/dev-router.php'],cwd=ROOT,env=self.env,stdout=self.log,stderr=self.log)
|
|
self.client=urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
for _ in range(100):
|
|
try:self.request();break
|
|
except urllib.error.URLError:time.sleep(.1)
|
|
else:self.close();raise RuntimeError('Test HTTP server did not start.')
|
|
def fixture(self,action,*extra):
|
|
args=PHP+['tests/http_fixture.php',action]+([self.database] if hasattr(self,'database') else [])+list(extra)
|
|
result=subprocess.run(args,cwd=ROOT,env=self.env,capture_output=True,text=True)
|
|
if result.returncode:raise RuntimeError(result.stdout+result.stderr)
|
|
return json.loads(result.stdout.splitlines()[-1]) if result.stdout.strip() else {}
|
|
def request(self,path='/',data=None,ajax=False):
|
|
req=urllib.request.Request(self.base+path,data=urllib.parse.urlencode(data).encode() if data is not None else None,headers={'X-Requested-With':'XMLHttpRequest'} if ajax else {})
|
|
try:
|
|
with self.client.open(req,timeout=30) as response:return response.status,response.read(),response.headers
|
|
except urllib.error.HTTPError as error:return error.code,error.read(),error.headers
|
|
@staticmethod
|
|
def csrf(body):
|
|
match=re.search(rb'name="csrf(?:_token|-token)"[^>]*(?:value|content)="([a-f0-9]+)"',body)
|
|
if not match:raise AssertionError('Missing CSRF token: '+body[:500].decode(errors='replace'))
|
|
return match.group(1).decode()
|
|
def login(self,email='admin@example.test',password='Admin test password 123!'):
|
|
_,body,_=self.request()
|
|
status,body,_=self.request(data={'action':'SignIn.login','csrf_token':self.csrf(body),'user_email':email,'user_password':password})
|
|
if status!=200 or b'name="csrf-token"' not in body:raise AssertionError('Login failed: '+body[:500].decode(errors='replace'))
|
|
if b'<h1>Account security</h1>' in body:
|
|
if b'setup key' in body:
|
|
secret=re.search(rb'<pre>([A-Z2-7]+)</pre>',body).group(1).decode()
|
|
code=self.totp(secret)
|
|
status,body,_=self.request(data={'action':'security','csrf_token':self.csrf(body),'current_password':password,'mfa_code':code})
|
|
if b'Save your recovery codes' not in body:raise AssertionError(body[:1000])
|
|
recovery=re.findall(rb'\b[a-f0-9]{32}\b',body)
|
|
self.recovery=[c.decode() for c in recovery]
|
|
status,body,_=self.request()
|
|
else:
|
|
code=self.fixture('mfaCode',email)['code']
|
|
status,body,_=self.request(data={'action':'security','csrf_token':self.csrf(body),'current_password':password,'mfa_code':code})
|
|
if status!=200:raise AssertionError('MFA failed: '+str(status))
|
|
self.token=self.csrf(body)
|
|
@staticmethod
|
|
def totp(secret, at=None):
|
|
import base64,hashlib,hmac,struct,time
|
|
secret=base64.b32decode(secret+'='*((8-len(secret)%8)%8))
|
|
digest=hmac.new(secret,struct.pack('>Q',int((time.time() if at is None else at)//30)),hashlib.sha1).digest()
|
|
offset=digest[-1]&15
|
|
return str((int.from_bytes(digest[offset:offset+4],'big')&0x7fffffff)%1000000).zfill(6)
|
|
def post(self,action,**values):
|
|
return self.request(data={'action':action,'csrf_token':self.token,**values},ajax=True)
|
|
def logout(self):self.post('signout')
|
|
def close(self):
|
|
self.server.terminate();self.server.wait(timeout=10)
|
|
self.fixture('drop')
|
|
self.log.seek(0);folder=ROOT/'tests/.local';folder.mkdir(exist_ok=True);(folder/'last-http.log').write_bytes(self.log.read());self.log.close()
|