108 lines
5.0 KiB
Python
108 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the running PHP workbench without cookies or security headers."""
|
|
import argparse
|
|
import json
|
|
import math
|
|
import re
|
|
import struct
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
class Client:
|
|
def __init__(self, base):
|
|
self.base = base.rstrip('/') + '/'
|
|
self.opener = urllib.request.build_opener(NoRedirect())
|
|
self.count = 0
|
|
|
|
def request(self, query='', data=None, binary=False, status=200):
|
|
self.count += 1
|
|
assert self.count <= 45, 'Request limit exceeded.'
|
|
headers = {}
|
|
if data is not None:
|
|
headers['Content-Type'] = 'application/octet-stream' if binary else 'application/json'
|
|
if not binary:
|
|
data = json.dumps(data).encode()
|
|
request = urllib.request.Request(self.base + query, data=data, headers=headers)
|
|
try:
|
|
response = self.opener.open(request, timeout=8)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
with response:
|
|
body = response.read(1048577)
|
|
assert len(body) <= 1048576, 'Response exceeds test read limit.'
|
|
assert response.code == status, (response.code, body[:300])
|
|
assert response.geturl() == request.full_url, 'Unexpected redirect.'
|
|
assert not response.headers.get('Set-Cookie'), 'Unexpected session cookie.'
|
|
return body, response.headers
|
|
|
|
def api(self, operation, query=None, data=None, binary=False, status=200):
|
|
params = {'api': operation, **(query or {})}
|
|
return self.request('?' + urllib.parse.urlencode(params), data, binary, status)
|
|
|
|
def json(self, operation, **kwargs):
|
|
raw, _ = self.api(operation, **kwargs)
|
|
value = json.loads(raw)
|
|
assert value.get('ok') is True, value
|
|
return value['data']
|
|
|
|
|
|
def run(base):
|
|
client = Client(base)
|
|
html, _ = client.request()
|
|
match = re.search(rb'window\.EXOPLASIM\s*=\s*(\{.*?\})\s*;', html, re.S)
|
|
assert match and set(json.loads(match[1])) == {'api'}, 'Unexpected bootstrap/session data.'
|
|
schema = client.json('schema')
|
|
assert schema['fields'] and schema['map_fields']
|
|
client.api('unknown', status=404)
|
|
suffix = uuid.uuid4().hex[:12]
|
|
fixtures, versions = [], []
|
|
try:
|
|
saved = client.json('experiments', data={'name': 'HTTP-' + suffix, 'spec': {'model': {}, 'run': {}, 'parameters': {}}})
|
|
fixtures.append(saved['id'])
|
|
clone = client.json('experiments', data={'name': 'HTTP-copy-' + suffix, 'spec': saved['spec']})
|
|
fixtures.append(clone['id'])
|
|
exported = json.loads(json.dumps({'name': 'HTTP-import-' + suffix, 'spec': clone['spec']}))
|
|
imported = client.json('experiments', data=exported)
|
|
fixtures.append(imported['id'])
|
|
assert imported['spec'] == saved['spec']
|
|
client.api('experiments', data={'id': saved['id'], 'name': saved['name'], 'spec': {'invalid': True}}, status=422)
|
|
retained = next(row for row in client.json('experiments') if row['id'] == saved['id'])
|
|
assert retained == saved, 'Failed save changed the stored preset.'
|
|
resolution = next(iter(schema['map_grids']))
|
|
grid = schema['map_grids'][resolution]
|
|
count = grid['nlat'] * grid['nlon']
|
|
payload = struct.pack('<f', 100.0) * count
|
|
query = {'field': 'height', 'resolution': resolution, 'name': 'HTTP map ' + suffix}
|
|
for _ in range(2):
|
|
versions.append(client.json('maps', query=query, data=payload, binary=True, status=201)['id'])
|
|
assert versions[0] != versions[1]
|
|
raw, headers = client.api('map', query={'id': versions[0]})
|
|
assert raw == payload and headers['X-Map-Shape'] == f"{grid['nlat']},{grid['nlon']}"
|
|
for invalid in [payload[:-1], struct.pack('<f', math.nan) * count, struct.pack('<f', schema['map_fields']['height']['max'] + 1) * count]:
|
|
client.api('maps', query=query, data=invalid, binary=True, status=422)
|
|
client.api('maps', query={**query, 'field': 'landmask'}, data=struct.pack('<f', .5) * count, binary=True, status=422)
|
|
client.api('maps', query={'resolution': 'invalid'}, status=422)
|
|
client.api('map', query={'id': '../invalid'}, status=404)
|
|
listed = client.json('maps', query={'resolution': resolution})['maps']
|
|
assert set(versions) <= {row['id'] for row in listed}
|
|
finally:
|
|
for identifier in fixtures:
|
|
assert client.json('experiment_delete', data={'id': identifier}) is True
|
|
assert not set(fixtures) & {row['id'] for row in client.json('experiments')}
|
|
return {'requests': client.count, 'deleted_presets': fixtures, 'saved_map_versions': versions}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--url', required=True)
|
|
arguments = parser.parse_args()
|
|
print(json.dumps({'ok': True, **run(arguments.url)}))
|