File size: 6,345 Bytes
c61c435 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | from __future__ import annotations
import http.client
import json
from pathlib import Path
import pytest
from adam.config import ConfigManager
from adam.remote_access import RemoteAccessService
from adam.remote_api import RemoteApiError, bounded_float
@pytest.fixture
def remote(tmp_path):
config = ConfigManager(tmp_path)
service = RemoteAccessService(config, None, None)
service.save_settings({'enabled': True})
# Let the OS allocate a port, avoiding a bind/close/rebind race in the test.
config.settings['remote_access']['port'] = 0
service.start()
try:
yield service
finally:
service.shutdown()
def request(service, path='/api/status', *, method='GET', body=None, headers=None, token=None):
connection = http.client.HTTPConnection('127.0.0.1', service._server.server_port, timeout=3)
auth = token if token is not None else service.settings()['token']
fields = {'Authorization': f'Bearer {auth}'}
fields.update(headers or {})
try:
connection.request(method, path, body=body, headers=fields)
response = connection.getresponse()
return response.status, dict(response.getheaders()), response.read()
finally:
connection.close()
def test_token_rotation_and_disable_take_effect_without_restart(remote):
old = remote.settings()['token']
assert request(remote, token=old)[0] == 200
remote.save_settings({'token': 'replacement-token'})
assert request(remote, token=old)[0] == 401
assert request(remote)[0] == 200
remote.save_settings({'enabled': False})
assert request(remote)[0] == 401
def test_missing_token_is_persisted_once(tmp_path):
config = ConfigManager(tmp_path)
config.update({'remote_access': {'enabled': False}})
service = RemoteAccessService(config, None, None)
try:
first = service.settings()['token']
assert service.settings()['token'] == first
assert ConfigManager(tmp_path).get('remote_access')['token'] == first
finally:
service.shutdown()
def test_remote_cannot_grant_its_own_control(remote):
payload = json.dumps({'auto_approve_training': True})
args = dict(method='POST', body=payload, headers={'Content-Type': 'application/json'})
assert request(remote, '/api/remote-settings', **args)[0] == 403
assert remote.settings()['auto_approve_training'] is False
remote.save_settings({'allow_job_control': True})
assert request(remote, '/api/remote-settings', **args)[0] == 200
remote.save_settings({'allow_job_control': False})
args['body'] = json.dumps({'auto_approve_training': False})
assert request(remote, '/api/remote-settings', **args)[0] == 200
@pytest.mark.parametrize('headers,body,status', [
({'Content-Type': 'application/json', 'Origin': 'https://attacker.invalid'}, '{}', 403),
({'Content-Type': 'application/json', 'Sec-Fetch-Site': 'cross-site'}, '{}', 403),
({'Content-Type': 'text/plain'}, '{}', 415),
({'Content-Type': 'application/json', 'Content-Length': '-1'}, '', 400),
({'Content-Type': 'application/json', 'Content-Length': '20001'}, '', 413),
({'Content-Type': 'application/json'}, '[]', 400),
({'Content-Type': 'application/json'}, '{', 400),
({'Content-Type': 'application/json'}, '{"auto_approve_training":"false"}', 400),
])
def test_rejects_unsafe_requests_before_mutation(remote, headers, body, status):
assert request(remote, '/api/remote-settings', method='POST', body=body, headers=headers)[0] == status
assert remote.settings()['auto_approve_training'] is False
def test_browser_security_headers(remote):
status, headers, _ = request(remote, '/')
assert status == 200
assert headers['Referrer-Policy'] == 'no-referrer'
assert headers['X-Frame-Options'] == 'DENY'
assert "frame-ancestors 'none'" in headers['Content-Security-Policy']
@pytest.mark.parametrize('value', ['nan', 'inf', '-inf', float('nan')])
def test_nonfinite_remote_settings_are_rejected(value):
with pytest.raises(RemoteApiError):
bounded_float(value, minimum=0, maximum=10, default=1, label='Guidance')
def test_dashboard_treats_remote_labels_as_text(tmp_path):
import shutil
import subprocess
from adam.remote_dashboard import remote_dashboard_app_html
node = shutil.which('node')
if not node:
pytest.skip('Node is needed for the dashboard JavaScript regression check')
html = remote_dashboard_app_html()
script = html.split('<script>', 1)[1].split('</script>', 1)[0]
names = ['$', 'list', 'clear', 'text', 'appendText', 'renderQueues', 'renderSystem', 'renderLocations', 'makeSetting', 'fieldId']
functions = '\n'.join(line for line in script.splitlines() if any(line.startswith('function ' + name + '(') for name in names))
harness = r'''
const assert = require('assert');
class Element {
constructor(tag) { this.tagName=tag; this.children=[]; this.textContent=''; }
set innerHTML(value) { throw new Error('Untrusted text reached HTML parsing'); }
appendChild(child) { this.children.push(child); }
get firstChild() { return this.children[0]; }
removeChild(child) { this.children.splice(this.children.indexOf(child),1); }
cloneNode() { return this; }
setAttribute() {}
}
const elements={};
const document={createElement:tag=>new Element(tag),getElementById:id=>elements[id]||(elements[id]=new Element('div'))};
const attack='<img src=x onerror="globalThis.compromised=true">';
var state={locationFilter:'',locations:[{id:'1',name:attack,source:attack,available:true}]};
'''
checks = r'''
renderQueues({queue:[{project:attack,status:attack,progress:0}]});
assert.equal(elements.queues.children[0].children[0].textContent,attack);
renderLocations();
assert.equal(elements.locationsList.children[0].children[0].textContent,attack);
makeSetting('preview',{type:'bool',label:attack},false);
renderSystem({cpu_percent:attack});
assert.equal(globalThis.compromised,undefined);
'''
# Parse the complete shipped script too, not just the rendering functions.
path = tmp_path / 'dashboard-test.js'
path.write_text('new Function(' + json.dumps(script) + ');\n' + harness + functions + checks, encoding='utf-8')
result = subprocess.run([node, str(path)], capture_output=True, text=True, timeout=10)
assert result.returncode == 0, result.stderr
|