tests/test_remote_hardening.py

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