sashatrask
279 lines
11 KiB
Python
279 lines
11 KiB
Python
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
TESTS_DIR = ROOT / 'tests'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
sys.path.insert(0, str(TESTS_DIR))
|
|
|
|
try:
|
|
from playwright.sync_api import sync_playwright
|
|
except ImportError:
|
|
sync_playwright = None
|
|
|
|
import uvicorn
|
|
|
|
from admin_api import EDGE_MARKER_HEADER, OPERATOR_HEADER, SECURITY_HEADERS
|
|
import test_admin_api as admin_test_fixture
|
|
from worker_api import create_worker_app
|
|
|
|
|
|
PUBLIC_PREFIX = '/browser-fixture-prefix'
|
|
|
|
|
|
def _browser_executable():
|
|
configured = os.environ.get('TRUF_BROWSER_EXECUTABLE')
|
|
candidates = [
|
|
configured,
|
|
shutil.which('google-chrome'),
|
|
shutil.which('chromium'),
|
|
shutil.which('chromium-browser'),
|
|
shutil.which('chrome'),
|
|
r'C:\Program Files\Google\Chrome\Application\chrome.exe',
|
|
r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe',
|
|
]
|
|
return next((value for value in candidates if value and os.path.isfile(value)), None)
|
|
|
|
|
|
BROWSER_EXECUTABLE = _browser_executable()
|
|
|
|
|
|
class _PrefixAdapter:
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope['type'] != 'http' or not (
|
|
scope['path'] == PUBLIC_PREFIX
|
|
or scope['path'].startswith(PUBLIC_PREFIX + '/')
|
|
):
|
|
await self.app(scope, receive, send)
|
|
return
|
|
suffix = scope['path'][len(PUBLIC_PREFIX):]
|
|
mapped = dict(scope)
|
|
mapped['path'] = '/admin-internal' + suffix
|
|
mapped['raw_path'] = mapped['path'].encode('ascii')
|
|
await self.app(mapped, receive, send)
|
|
|
|
|
|
class AdminBrowserTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if sync_playwright is None:
|
|
raise RuntimeError(
|
|
'Playwright is required; install tests/requirements-browser.txt'
|
|
)
|
|
cls.fixture = admin_test_fixture.AdminAPITests(methodName='runTest')
|
|
cls.fixture.setUp()
|
|
app = create_worker_app(
|
|
cls.fixture.worker, reaper_interval_seconds=3600,
|
|
admin_service=cls.fixture.admin,
|
|
)
|
|
cls.app = _PrefixAdapter(app)
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
|
probe.bind(('127.0.0.1', 0))
|
|
cls.port = probe.getsockname()[1]
|
|
config = uvicorn.Config(
|
|
cls.app, host='127.0.0.1', port=cls.port,
|
|
log_level='critical', lifespan='off', access_log=False,
|
|
)
|
|
cls.server = uvicorn.Server(config)
|
|
cls.thread = threading.Thread(target=cls.server.run, daemon=True)
|
|
cls.thread.start()
|
|
deadline = time.monotonic() + 10
|
|
while not cls.server.started and cls.thread.is_alive():
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError('browser fixture server did not start')
|
|
time.sleep(0.02)
|
|
if not cls.server.started:
|
|
raise RuntimeError('browser fixture server failed')
|
|
cls.base_url = f'http://127.0.0.1:{cls.port}{PUBLIC_PREFIX}'
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.should_exit = True
|
|
cls.thread.join(timeout=10)
|
|
cls.fixture.tearDown()
|
|
|
|
def _assert_storage_empty(self, page):
|
|
state = page.evaluate('''async () => ({
|
|
local: Object.keys(localStorage),
|
|
session: Object.keys(sessionStorage),
|
|
caches: 'caches' in self ? await caches.keys() : [],
|
|
serviceWorkers: 'serviceWorker' in navigator
|
|
? (await navigator.serviceWorker.getRegistrations()).length : 0,
|
|
indexedDb: indexedDB.databases
|
|
? (await indexedDB.databases()).map(item => item.name || '') : [],
|
|
})''')
|
|
self.assertEqual(state, {
|
|
'local': [], 'session': [], 'caches': [],
|
|
'serviceWorkers': 0, 'indexedDb': [],
|
|
})
|
|
|
|
def _assert_page_contract(self, page, response, *, script_count=0):
|
|
self.assertIsNotNone(response)
|
|
self.assertEqual(response.status, 200)
|
|
headers = response.all_headers()
|
|
for name, expected in SECURITY_HEADERS.items():
|
|
self.assertEqual(headers.get(name.lower()), expected)
|
|
self.assertEqual(page.locator('script').count(), script_count)
|
|
if script_count:
|
|
self.assertEqual(
|
|
page.locator('script:not([src])').count(), 0,
|
|
)
|
|
event_attributes = page.locator('*').evaluate_all('''elements =>
|
|
elements.flatMap(element => Array.from(element.attributes))
|
|
.filter(attribute => attribute.name.toLowerCase().startsWith('on'))
|
|
.map(attribute => attribute.name)
|
|
''')
|
|
self.assertEqual(event_attributes, [])
|
|
scroll_width = page.evaluate('document.documentElement.scrollWidth')
|
|
client_width = page.evaluate('document.documentElement.clientWidth')
|
|
overflowing = page.locator('*').evaluate_all('''elements => elements
|
|
.filter(element => element.getBoundingClientRect().right >
|
|
document.documentElement.clientWidth + 1)
|
|
.slice(0, 10)
|
|
.map(element => ({
|
|
tag: element.tagName,
|
|
className: element.className,
|
|
right: element.getBoundingClientRect().right,
|
|
scrollWidth: element.scrollWidth,
|
|
display: getComputedStyle(element).display,
|
|
width: getComputedStyle(element).width,
|
|
parent: element.parentElement && {
|
|
tag: element.parentElement.tagName,
|
|
className: element.parentElement.className,
|
|
width: getComputedStyle(element.parentElement).width,
|
|
overflow: getComputedStyle(element.parentElement).overflowX,
|
|
},
|
|
}))''')
|
|
self.assertLessEqual(scroll_width, client_width, overflowing)
|
|
for target in page.locator('a[href], link[href], form[action]').evaluate_all(
|
|
'''elements => elements.map(element =>
|
|
new URL(element.getAttribute('href') || element.getAttribute('action'),
|
|
document.baseURI).href)'''
|
|
):
|
|
parsed = urlparse(target)
|
|
self.assertEqual(parsed.netloc, f'127.0.0.1:{self.port}')
|
|
self.assertTrue(parsed.path.startswith(PUBLIC_PREFIX))
|
|
|
|
def test_navigation_csp_mobile_and_secret_non_retention(self):
|
|
console_errors = []
|
|
with sync_playwright() as playwright:
|
|
launch_options = {'headless': True, 'args': ['--disable-gpu']}
|
|
if BROWSER_EXECUTABLE is not None:
|
|
launch_options['executable_path'] = BROWSER_EXECUTABLE
|
|
browser = playwright.chromium.launch(**launch_options)
|
|
context = browser.new_context(
|
|
viewport={'width': 1440, 'height': 900},
|
|
extra_http_headers={
|
|
EDGE_MARKER_HEADER: admin_test_fixture.MARKER,
|
|
OPERATOR_HEADER: admin_test_fixture.OPERATOR,
|
|
},
|
|
)
|
|
context.grant_permissions(
|
|
['clipboard-read', 'clipboard-write'],
|
|
origin=f'http://127.0.0.1:{self.port}',
|
|
)
|
|
page = context.new_page()
|
|
page.on(
|
|
'console',
|
|
lambda message: console_errors.append(message.text)
|
|
if message.type == 'error' else None,
|
|
)
|
|
|
|
response = page.goto(self.base_url + '/secrets', wait_until='load')
|
|
self._assert_page_contract(page, response)
|
|
textarea = page.locator('textarea[name="document_text"]')
|
|
initial_secret = textarea.input_value()
|
|
self.assertIn('editor-secret', initial_secret)
|
|
self.assertEqual(page.content().count('editor-secret'), 1)
|
|
self.assertNotIn(
|
|
'editor-secret',
|
|
page.locator('body').evaluate('''body => {
|
|
const clone = body.cloneNode(true);
|
|
clone.querySelectorAll('textarea').forEach(item => item.remove());
|
|
return clone.textContent;
|
|
}'''),
|
|
)
|
|
self.assertEqual(textarea.get_attribute('autocomplete'), 'off')
|
|
self.assertEqual(textarea.locator('xpath=ancestor::form').get_attribute('autocomplete'), 'off')
|
|
self._assert_storage_empty(page)
|
|
|
|
unsaved = 'browser-unsaved-secret-must-not-persist'
|
|
textarea.fill(initial_secret + '\n' + unsaved)
|
|
page.goto(self.base_url + '/overview', wait_until='load')
|
|
page.go_back(wait_until='load')
|
|
page.reload(wait_until='load')
|
|
self.assertNotIn(unsaved, textarea.input_value())
|
|
self.assertNotIn(unsaved, page.content())
|
|
self._assert_storage_empty(page)
|
|
|
|
response = page.goto(self.base_url + '/', wait_until='load')
|
|
self._assert_page_contract(page, response)
|
|
response = page.goto(
|
|
self.base_url + '/assignments/5', wait_until='load',
|
|
)
|
|
self._assert_page_contract(page, response, script_count=1)
|
|
copy_button = page.locator('[data-copy-target]').first
|
|
expected_json = page.locator('.canonical-json').first.input_value()
|
|
copy_button.click()
|
|
deadline = time.monotonic() + 5
|
|
while (
|
|
copy_button.text_content() != 'Copied canonical JSON'
|
|
and time.monotonic() < deadline
|
|
):
|
|
page.wait_for_timeout(20)
|
|
self.assertEqual(copy_button.text_content(), 'Copied canonical JSON')
|
|
self.assertEqual(
|
|
page.evaluate('navigator.clipboard.readText()'), expected_json,
|
|
)
|
|
self._assert_storage_empty(page)
|
|
|
|
response = page.goto(self.base_url + '/', wait_until='load')
|
|
self._assert_page_contract(page, response)
|
|
navigation = page.locator('nav a').evaluate_all(
|
|
'links => links.map(link => new URL(link.href).href)'
|
|
)
|
|
self.assertGreaterEqual(len(navigation), 8)
|
|
for target in navigation:
|
|
self.assertTrue(urlparse(target).path.startswith(PUBLIC_PREFIX))
|
|
response = page.goto(target, wait_until='load')
|
|
self._assert_page_contract(page, response)
|
|
|
|
self.assertEqual(console_errors, [])
|
|
page.goto(self.base_url + '/overview', wait_until='load')
|
|
page.evaluate('''() => {
|
|
const probe = document.createElement('script');
|
|
probe.textContent = 'window.__trufInlineCspProbe = true';
|
|
document.body.appendChild(probe);
|
|
}''')
|
|
page.wait_for_timeout(100)
|
|
self.assertFalse(page.evaluate('Boolean(window.__trufInlineCspProbe)'))
|
|
|
|
page.set_viewport_size({'width': 390, 'height': 844})
|
|
for path in ('/supervisor', '/files', '/audit', '/secrets'):
|
|
response = page.goto(self.base_url + path, wait_until='load')
|
|
self._assert_page_contract(page, response)
|
|
response = page.goto(
|
|
self.base_url + '/assignments/5', wait_until='load',
|
|
)
|
|
self._assert_page_contract(page, response, script_count=1)
|
|
|
|
self._assert_storage_empty(page)
|
|
context.close()
|
|
browser.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|