diff --git a/requirements-ui-tests.txt b/requirements-ui-tests.txt new file mode 100644 index 000000000..c85b2afa6 --- /dev/null +++ b/requirements-ui-tests.txt @@ -0,0 +1,12 @@ +# Requirements for Playwright E2E UI tests +# Install with: pip install -r requirements-ui-tests.txt +# Then: playwright install chromium +# +# nodeenv is used by the tox `ui` environment to install Node.js inside the +# Python virtualenv so `npm run build` can compile the Magma frontend before +# tests run. + +playwright>=1.44.0 +pytest-playwright>=0.5.0 +requests>=2.31.0 +nodeenv>=1.8.0 diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/e2e/magma/__init__.py b/tests/e2e/magma/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/e2e/magma/conftest.py b/tests/e2e/magma/conftest.py new file mode 100644 index 000000000..74345f236 --- /dev/null +++ b/tests/e2e/magma/conftest.py @@ -0,0 +1,242 @@ +""" +Pytest configuration for Playwright E2E UI tests. + +Each test session finds a free TCP port, writes a temporary conf/local.yml +(so multiple VENVs / CI jobs can run simultaneously without port conflicts), +starts Caldera on that port, waits until healthy, provides an authenticated +Playwright page, then tears everything down. + +Prerequisites (installed by the `ui` tox environment): + pip install playwright pytest-playwright requests + playwright install chromium + +Run: + pytest plugins/magma/tests/e2e -v --browser chromium + +Environment variables: + CALDERA_PORT Force a specific port (default: auto-detect a free port) + CALDERA_USER Username to log in as (default: admin) + CALDERA_PASS Password (default: admin) + CALDERA_EXTERNAL Set to '1' to skip server startup and use CALDERA_URL + CALDERA_URL Base URL when CALDERA_EXTERNAL=1 + CALDERA_STARTUP_TIMEOUT Seconds to wait for server (default: 90) +""" + +import os +import shutil +import socket +import subprocess +import tempfile +import time + +import pytest +import requests +import yaml + +# --------------------------------------------------------------------------- +# Paths +# --------------------------------------------------------------------------- + +_HERE = os.path.dirname(__file__) +CALDERA_ROOT = os.path.normpath(os.path.join(_HERE, '..', '..', '..', '..', '..')) +CONF_DIR = os.path.join(CALDERA_ROOT, 'conf') +DEFAULT_YML = os.path.join(CONF_DIR, 'default.yml') +LOCAL_YML = os.path.join(CONF_DIR, 'local.yml') + +CALDERA_USER = os.environ.get('CALDERA_USER', 'admin') +CALDERA_PASS = os.environ.get('CALDERA_PASS', 'admin') +STARTUP_TIMEOUT = int(os.environ.get('CALDERA_STARTUP_TIMEOUT', '90')) + + +# --------------------------------------------------------------------------- +# Port helpers +# --------------------------------------------------------------------------- + +def _find_free_port() -> int: + """Bind to port 0 to let the OS assign a free ephemeral port.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('127.0.0.1', 0)) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + return s.getsockname()[1] + + +def _port_in_use(port: int) -> bool: + """Return True if something is already listening on *port*.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(1) + return s.connect_ex(('127.0.0.1', port)) == 0 + + +# --------------------------------------------------------------------------- +# Config helpers +# --------------------------------------------------------------------------- + +def _write_test_local_yml(port: int) -> None: + """ + Write conf/local.yml based on conf/default.yml but with: + - port overridden to *port* + - host set to 127.0.0.1 (test-only, no external exposure) + + If conf/local.yml already exists it is backed up first so the original + is restored in teardown. + """ + with open(DEFAULT_YML, 'r', encoding='utf-8') as fh: + config = yaml.safe_load(fh) + + config['port'] = port + config['host'] = '127.0.0.1' + + with open(LOCAL_YML, 'w', encoding='utf-8') as fh: + yaml.safe_dump(config, fh, default_flow_style=False) + + +# --------------------------------------------------------------------------- +# Server lifecycle +# --------------------------------------------------------------------------- + +@pytest.fixture(scope='session') +def caldera_server(): + """ + Start Caldera on a free port using conf/local.yml, wait until healthy, + yield the base URL, then terminate and clean up. + + Set CALDERA_EXTERNAL=1 to skip startup and connect to an already-running + instance at CALDERA_URL instead. + """ + if os.environ.get('CALDERA_EXTERNAL') == '1': + yield os.environ.get('CALDERA_URL', 'http://localhost:8888') + return + + # Choose port — env var overrides auto-detection + port = int(os.environ.get('CALDERA_PORT', _find_free_port())) + base_url = f'http://127.0.0.1:{port}' + + # Safety: refuse to start if something is already on this port. + # Caldera loads all state into memory; writing conf/local.yml while a + # running instance holds config in memory would be silently ignored and + # could leave the config file in an inconsistent state. + if _port_in_use(port): + pytest.fail( + f'Port {port} is already in use. ' + 'Stop any running Caldera instance before running UI tests, ' + 'or set CALDERA_PORT to a free port.' + ) + + # Back up existing local.yml if present + local_yml_backup = None + if os.path.exists(LOCAL_YML): + local_yml_backup = LOCAL_YML + '.e2e_backup' + shutil.copy2(LOCAL_YML, local_yml_backup) + + _write_test_local_yml(port) + + env = os.environ.copy() + env['PYTHONPATH'] = CALDERA_ROOT + + proc = subprocess.Popen( + # -E local → uses conf/local.yml we just wrote + # -l ERROR → suppress startup noise in test output + ['python', 'server.py', '-E', 'local', '-l', 'ERROR'], + cwd=CALDERA_ROOT, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + health_url = f'{base_url}/api/v2/health' + deadline = time.time() + STARTUP_TIMEOUT + ready = False + while time.time() < deadline: + try: + r = requests.get(health_url, timeout=2) + if r.status_code == 200: + ready = True + break + except requests.RequestException: + pass + time.sleep(1) + + if not ready: + proc.terminate() + # Restore backup before failing + if local_yml_backup: + shutil.move(local_yml_backup, LOCAL_YML) + else: + os.remove(LOCAL_YML) + pytest.fail( + f'Caldera did not become healthy within {STARTUP_TIMEOUT}s ' + f'(checked {health_url}). ' + f'Port {port} was selected.' + ) + + yield base_url + + # Teardown + proc.terminate() + try: + proc.wait(timeout=15) + except subprocess.TimeoutExpired: + proc.kill() + + if local_yml_backup: + shutil.move(local_yml_backup, LOCAL_YML) + elif os.path.exists(LOCAL_YML): + os.remove(LOCAL_YML) + + +# --------------------------------------------------------------------------- +# API session (requests) +# --------------------------------------------------------------------------- + +@pytest.fixture(scope='session') +def api_session(caldera_server): + """ + Authenticated requests.Session for direct API calls in tests. + Used to set up / verify data independently of the browser. + """ + session = requests.Session() + resp = session.post( + f'{caldera_server}/enter', + data={'username': CALDERA_USER, 'password': CALDERA_PASS}, + allow_redirects=False, + ) + assert resp.status_code in (200, 302), ( + f'Login failed: HTTP {resp.status_code}. ' + f'Verify CALDERA_USER/CALDERA_PASS match conf/local.yml credentials.' + ) + return session + + +@pytest.fixture(scope='session') +def base_url(caldera_server): + """Base URL of the running Caldera instance (e.g. http://127.0.0.1:54321).""" + return caldera_server + + +# --------------------------------------------------------------------------- +# Playwright fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def auth_page(page, caldera_server, api_session): + """ + Playwright ``page`` pre-loaded with authentication cookies. + + Copies the requests.Session cookies into the Playwright browser context + so tests start already logged in without going through the login form. + The page is NOT yet navigated — tests must call ``page.goto(...)`` first. + """ + pw_cookies = [ + { + 'name': c.name, + 'value': c.value, + 'domain': '127.0.0.1', + 'path': '/', + 'httpOnly': False, + 'secure': False, + } + for c in api_session.cookies + ] + if pw_cookies: + page.context.add_cookies(pw_cookies) + yield page diff --git a/tests/e2e/magma/test_abilities.py b/tests/e2e/magma/test_abilities.py new file mode 100644 index 000000000..ce01c886a --- /dev/null +++ b/tests/e2e/magma/test_abilities.py @@ -0,0 +1,264 @@ +""" +E2E tests for the Caldera Abilities page (AbilitiesView.vue). + +These tests cover page structure, filter controls, the ability list, and the +create-ability modal. All tests use the `auth_page` fixture so auth cookies +are already present before navigation. + +Run with: + pytest plugins/magma/tests/e2e/test_abilities.py -v --browser chromium +""" + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# 1. h2 heading "Abilities" is visible +# --------------------------------------------------------------------------- + +def test_abilities_page_heading(auth_page: Page, base_url: str) -> None: + """ + Navigating to /abilities must render an

whose text is "Abilities". + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + heading = auth_page.locator('h2', has_text='Abilities') + expect(heading).to_be_visible() + + +# --------------------------------------------------------------------------- +# 2. Introductory ATT&CK description paragraph is visible +# --------------------------------------------------------------------------- + +def test_abilities_page_description(auth_page: Page, base_url: str) -> None: + """ + The page must display a descriptive paragraph that mentions ATT&CK — the + text matches the static copy rendered just below the

. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + # The paragraph contains the phrase used in the Vue template + description = auth_page.locator('p', has_text='ATT&CK') + expect(description).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. "Create an Ability" button is visible +# --------------------------------------------------------------------------- + +def test_create_ability_button_visible(auth_page: Page, base_url: str) -> None: + """ + The primary call-to-action button labelled "Create an Ability" must be + present and visible in the left sidebar column. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + create_btn = auth_page.locator('button', has_text='Create an Ability') + expect(create_btn).to_be_visible() + + +# --------------------------------------------------------------------------- +# 4. Search input with placeholder "Find an ability..." is visible +# --------------------------------------------------------------------------- + +def test_search_input_visible(auth_page: Page, base_url: str) -> None: + """ + A text input with placeholder "Find an ability..." must be rendered in the + filter sidebar so users can search the ability list by name. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + search_input = auth_page.locator('input[placeholder="Find an ability..."]') + expect(search_input).to_be_visible() + + +# --------------------------------------------------------------------------- +# 5. Tactic filter dropdown is visible and contains an "All" option +# --------------------------------------------------------------------------- + +def test_tactic_filter_dropdown_visible(auth_page: Page, base_url: str) -> None: + """ + The Tactic filter must be rendered as a or a Bulma dropdown; try element first; fall back to Bulma dropdown trigger + item. + select_el = selector_column.locator("select") + if select_el.count() > 0: + # Select the option whose text matches the first objective's name. + if first_name: + select_el.select_option(label=first_name) + else: + # If name is blank, select by index (skip placeholder at 0). + select_el.select_option(index=1) + else: + # Bulma dropdown: click the trigger to open, then click the first item. + dropdown_trigger = selector_column.locator(".dropdown-trigger button").first + dropdown_trigger.click() + first_item = selector_column.locator(".dropdown-item").first + expect(first_item).to_be_visible() + first_item.click() + + # Wait for the Vue reactive update to propagate. + auth_page.wait_for_load_state("networkidle") + + # The right-hand detail column should now display content. + detail_column = auth_page.locator(".column.is-9") + + # Assert the Save button is visible — it is rendered only when an objective + # is selected (controlled by Vue's v-if on the selected objective). + save_btn = detail_column.locator("button", has_text="Save") + expect(save_btn).to_be_visible() + + # If the objective has a name, it must be visible in the detail panel. + if first_name: + name_in_detail = detail_column.locator(f"text={first_name}") + expect(name_in_detail.first).to_be_visible() + + # The goals section (table or list) must also be present in the detail panel. + # Accept a , a , or an element labelled "goals". + goals_section = detail_column.locator( + "table, thead, [class*='goal'], th, td, text=goals" + ) + expect(goals_section.first).to_be_visible() diff --git a/tests/e2e/magma/test_operations.py b/tests/e2e/magma/test_operations.py new file mode 100644 index 000000000..5bfc436ef --- /dev/null +++ b/tests/e2e/magma/test_operations.py @@ -0,0 +1,212 @@ +""" +Playwright E2E tests for the Caldera Operations UI view (/operations). + +These tests verify the structure, visibility, and basic interactivity of +the Operations page in the Magma frontend (Vue 3 SPA). All tests are +independent and leave no permanent state — the operations list is observed +but never mutated. + +Fixtures used (provided by conftest.py): + caldera_server (session) — base URL string + api_session (session) — authenticated requests.Session + auth_page (function) — Playwright page with auth cookies, not yet navigated + base_url (function) — base URL string +""" + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def navigate_to_operations(page: Page, base_url: str) -> None: + """Navigate to the /operations route and wait for network activity to settle.""" + page.goto(f"{base_url}/operations") + page.wait_for_load_state("networkidle") + + +def get_operations_from_api(api_session, caldera_server: str) -> list: + """Return the list of operation objects from the REST API.""" + resp = api_session.get(f"{caldera_server}/api/v2/operations") + assert resp.status_code == 200, ( + f"GET /api/v2/operations returned HTTP {resp.status_code}" + ) + return resp.json() + + +def open_operation_dropdown(page: Page) -> None: + """ + Click (or hover) the operation selector dropdown button to expand it. + + The selector button in OperationsView is a Bulma dropdown; a click on the + trigger button exposes the dropdown-menu containing the operation list. + """ + # The primary selector button shows "Select an operation" when nothing is chosen + selector_btn = page.locator( + ".columns .column .is-flex .dropdown button.button.is-primary", + has_text="operation", + ).first + expect(selector_btn).to_be_visible() + selector_btn.hover() + # Wait for the dropdown menu to appear + page.wait_for_selector(".columns .column .is-flex .dropdown .dropdown-menu", state="visible", timeout=5000) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +def test_operations_page_heading(auth_page: Page, base_url: str) -> None: + """ + The

on the Operations page must contain the text "Operations". + This confirms the correct view is rendered by the Vue router. + """ + navigate_to_operations(auth_page, base_url) + + heading = auth_page.locator("h2") + expect(heading).to_be_visible() + expect(heading).to_contain_text("Operations") + + +def test_new_operation_button_visible(auth_page: Page, base_url: str) -> None: + """ + The "New Operation" primary button must be present and visible in the + toolbar area next to the operation selector. + """ + navigate_to_operations(auth_page, base_url) + + new_op_btn = auth_page.get_by_role("button", name="New Operation") + expect(new_op_btn).to_be_visible() + + +def test_operation_selector_dropdown_visible(auth_page: Page, base_url: str) -> None: + """ + The operation selector dropdown button must be present and visible, + initially showing the placeholder text "Select an operation". + """ + navigate_to_operations(auth_page, base_url) + + # The selector is a primary button that contains the placeholder text + # when no operation is currently selected. + selector_btn = auth_page.locator("button.button.is-primary", has_text="Select an operation") + expect(selector_btn).to_be_visible() + + +def test_new_operation_modal_opens(auth_page: Page, base_url: str) -> None: + """ + Clicking "New Operation" should open a modal overlay that contains at + minimum a text input field for the operation name. + + The Bulma modal becomes active by receiving the `is-active` class. + """ + navigate_to_operations(auth_page, base_url) + + new_op_btn = auth_page.get_by_role("button", name="New Operation") + expect(new_op_btn).to_be_visible() + new_op_btn.click() + + # The Bulma modal gains `is-active` when opened + modal = auth_page.locator(".modal.is-active") + try: + expect(modal).to_be_visible(timeout=5000) + except Exception: + # Fallback: accept any visible dialog/overlay containing an input + modal = auth_page.locator("[class*='modal'], [role='dialog']") + expect(modal).to_be_visible(timeout=3000) + + # The modal must contain at least one text input (operation name field) + name_input = modal.locator("input[type='text'], input:not([type])") + expect(name_input.first).to_be_visible() + + +def test_operations_api_count_matches_dropdown( + auth_page: Page, base_url: str, api_session, caldera_server: str +) -> None: + """ + The number of items shown in the operation selector dropdown must equal + the count returned by GET /api/v2/operations. + + Steps: + 1. Fetch operations from the API to get the expected count. + 2. Navigate to /operations. + 3. Expand the operation selector dropdown. + 4. Count the rendered dropdown items and assert equality. + """ + operations = get_operations_from_api(api_session, caldera_server) + expected_count = len(operations) + + navigate_to_operations(auth_page, base_url) + open_operation_dropdown(auth_page) + + # Each operation renders as an inside the menu + dropdown_items = auth_page.locator( + ".columns .column .is-flex .dropdown .dropdown-menu .dropdown-content a.dropdown-item" + ) + actual_count = dropdown_items.count() + + assert actual_count == expected_count, ( + f"Dropdown shows {actual_count} operation(s) but API returned {expected_count}" + ) + + +def test_operation_names_in_dropdown( + auth_page: Page, base_url: str, api_session, caldera_server: str +) -> None: + """ + When at least one operation exists, the first operation's name returned by + GET /api/v2/operations must appear as a visible item in the dropdown. + + If no operations exist the test is skipped. + """ + operations = get_operations_from_api(api_session, caldera_server) + if not operations: + pytest.skip("No operations registered — cannot verify dropdown names.") + + first_name = operations[0]["name"] + + navigate_to_operations(auth_page, base_url) + open_operation_dropdown(auth_page) + + # The operation's name should appear as a visible dropdown item + item = auth_page.locator( + ".columns .column .is-flex .dropdown .dropdown-menu .dropdown-content a.dropdown-item", + has_text=first_name, + ) + expect(item).to_be_visible() + + +def test_delete_button_hidden_when_no_operation_selected( + auth_page: Page, base_url: str +) -> None: + """ + When no operation is selected, the "Delete" button must not be visible. + + In the template the Delete button uses `v-if="selectedOperation.id"` so it + should be absent from the DOM (or hidden) when the selection is empty. + """ + navigate_to_operations(auth_page, base_url) + + # The Delete button must not be visible when nothing is selected. + # Using to_not_be_visible covers both the v-if removal from DOM and a + # CSS display:none scenario. + delete_btn = auth_page.get_by_role("button", name="Delete") + expect(delete_btn).not_to_be_visible() + + +def test_download_report_hidden_when_no_operation_selected( + auth_page: Page, base_url: str +) -> None: + """ + When no operation is selected, the "Download Report" button must not be + visible. + + In the template the button uses `v-if="selectedOperation.id"` so it + should be absent from the DOM (or hidden) when the selection is empty. + """ + navigate_to_operations(auth_page, base_url) + + # The Download Report button must not be visible when nothing is selected. + download_btn = auth_page.get_by_role("button", name="Download Report") + expect(download_btn).not_to_be_visible() diff --git a/tests/e2e/magma/test_payloads.py b/tests/e2e/magma/test_payloads.py new file mode 100644 index 000000000..392cba1c0 --- /dev/null +++ b/tests/e2e/magma/test_payloads.py @@ -0,0 +1,228 @@ +""" +E2E tests for the Caldera Payloads page (PayloadsView.vue). + +Covers page structure (heading, description, file upload control), list/API +count parity, payload name rendering, and an upload-then-verify lifecycle +that cleans up after itself. + +All tests use the `auth_page` fixture so auth cookies are present before +navigation. Tests that upload data delete the created payload via +`api_session` so the suite remains idempotent. + +Run with: + pytest plugins/magma/tests/e2e/test_payloads.py -v --browser chromium +""" + +import io + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + +_TEST_PAYLOAD_NAME = 'test_e2e_payload.txt' +_TEST_PAYLOAD_CONTENT = b'test' + + +# --------------------------------------------------------------------------- +# 1. h2 heading "Payloads" is visible +# --------------------------------------------------------------------------- + +def test_payloads_page_heading(auth_page: Page, base_url: str) -> None: + """ + Navigating to /payloads must render an

whose text is "Payloads". + """ + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + heading = auth_page.locator('h2', has_text='Payloads') + expect(heading).to_be_visible() + + +# --------------------------------------------------------------------------- +# 2. Introductory description paragraph is visible +# --------------------------------------------------------------------------- + +def test_payloads_description_visible(auth_page: Page, base_url: str) -> None: + """ + The page must display a descriptive paragraph below the heading that + explains the purpose of payloads (text mentions "abilities", matching + the copy in the Pug template: "Payloads are files that can be attached + to abilities..."). + """ + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + description = auth_page.locator('p', has_text='abilities') + expect(description).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. Payload list item count matches the API count +# --------------------------------------------------------------------------- + +def test_payloads_api_count_matches_ui( + auth_page: Page, base_url: str, api_session +) -> None: + """ + The number of payload rows rendered in the UI must equal the number of + payload name strings returned by GET /api/v2/payloads. + + When the API returns zero payloads the test passes trivially. + """ + resp = api_session.get(base_url + '/api/v2/payloads') + assert resp.status_code == 200, ( + f'GET /api/v2/payloads returned HTTP {resp.status_code}' + ) + api_payloads = resp.json() + api_count = len(api_payloads) + + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + # Each payload is rendered as a row in a table or as a list item. + # We target

elements inside a (table layout) as the primary + # selector, falling back to
  • elements if the view uses a list layout. + tbody_rows = auth_page.locator('tbody tr') + list_items = auth_page.locator('ul li, ol li') + + if api_count == 0: + # Neither selector should produce any visible results + assert tbody_rows.count() == 0 or list_items.count() == 0, ( + 'API returned 0 payloads but row/list elements are visible' + ) + return + + # Prefer tbody rows; fall back to list items + if tbody_rows.count() > 0: + expect(tbody_rows.first).to_be_visible() + ui_count = tbody_rows.count() + else: + expect(list_items.first).to_be_visible() + ui_count = list_items.count() + + assert ui_count == api_count, ( + f'UI shows {ui_count} payload rows but API returned {api_count} payloads' + ) + + +# --------------------------------------------------------------------------- +# 4. First payload name from the API appears on the page +# --------------------------------------------------------------------------- + +def test_payload_names_displayed( + auth_page: Page, base_url: str, api_session +) -> None: + """ + After navigating to /payloads, the filename of the first payload returned + by GET /api/v2/payloads must be visible somewhere on the page, confirming + that the Vue component has rendered data fetched from the API. + + Skipped when no payloads are present in the system. + """ + resp = api_session.get(base_url + '/api/v2/payloads') + assert resp.status_code == 200, ( + f'GET /api/v2/payloads returned HTTP {resp.status_code}' + ) + payloads = resp.json() + + if not payloads: + pytest.skip('No payloads available to verify against the UI') + + first_name = payloads[0] + + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + name_locator = auth_page.locator(f'text={first_name}') + expect(name_locator.first).to_be_visible() + + +# --------------------------------------------------------------------------- +# 5. File upload input (or upload dropzone / button) is present on the page +# --------------------------------------------------------------------------- + +def test_upload_input_present(auth_page: Page, base_url: str) -> None: + """ + The Payloads page must provide a mechanism for uploading files. This is + expected to be either an `` element, a visible button + labelled "Upload", or a dropzone area — at least one of these must be + present and attached to the DOM. + """ + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + # Primary selector: a standard file input + file_input = auth_page.locator('input[type="file"]') + upload_button = auth_page.locator('button', has_text='Upload') + dropzone = auth_page.locator('[class*="dropzone"], [class*="drop-zone"], [class*="upload"]') + + # At least one upload mechanism must be present in the DOM + has_file_input = file_input.count() > 0 + has_upload_button = upload_button.count() > 0 + has_dropzone = dropzone.count() > 0 + + assert has_file_input or has_upload_button or has_dropzone, ( + 'No file upload control found on /payloads: ' + 'expected input[type="file"], an "Upload" button, or a dropzone element' + ) + + +# --------------------------------------------------------------------------- +# 6. A payload uploaded via the API appears in the UI; cleaned up afterwards +# --------------------------------------------------------------------------- + +def test_payload_upload_via_api_appears_in_ui( + auth_page: Page, base_url: str, api_session +) -> None: + """ + A small text payload POSTed directly to POST /api/v2/payloads must appear + on the /payloads page after navigation, confirming that the Vue component + renders the full server-side payload list. + + The test payload is deleted via DELETE /api/v2/payloads/{name} after the + assertion so no state is left behind. + """ + # --- Ensure no leftover test payload from a previous run --- + pre_delete = api_session.delete(base_url + f'/api/v2/payloads/{_TEST_PAYLOAD_NAME}') + # 404 is acceptable here (payload didn't exist); anything else is unexpected + assert pre_delete.status_code in (200, 204, 404), ( + f'Pre-test cleanup DELETE returned HTTP {pre_delete.status_code}' + ) + + # --- Upload the test payload via the API --- + upload_resp = api_session.post( + base_url + '/api/v2/payloads', + files={ + 'file': ( + _TEST_PAYLOAD_NAME, + io.BytesIO(_TEST_PAYLOAD_CONTENT), + 'text/plain', + ) + }, + ) + assert upload_resp.status_code in (200, 201), ( + f'POST /api/v2/payloads returned HTTP {upload_resp.status_code}: ' + f'{upload_resp.text}' + ) + + try: + # --- Navigate to /payloads and verify the uploaded file is listed --- + auth_page.goto(base_url + '/payloads') + auth_page.wait_for_load_state('networkidle') + + payload_text = auth_page.locator(f'text={_TEST_PAYLOAD_NAME}') + expect(payload_text.first).to_be_visible() + + finally: + # --- Clean up: delete the test payload regardless of assertion outcome --- + del_resp = api_session.delete( + base_url + f'/api/v2/payloads/{_TEST_PAYLOAD_NAME}' + ) + assert del_resp.status_code in (200, 204), ( + f'Cleanup DELETE /api/v2/payloads/{_TEST_PAYLOAD_NAME} returned ' + f'HTTP {del_resp.status_code}' + ) diff --git a/tests/e2e/magma/test_planners.py b/tests/e2e/magma/test_planners.py new file mode 100644 index 000000000..94182fe12 --- /dev/null +++ b/tests/e2e/magma/test_planners.py @@ -0,0 +1,179 @@ +""" +E2E tests for the Caldera Planners UI view (/planners). + +Planners are built-in modules that decide which abilities a red team agent +should execute during an operation. The view is read-only — users cannot +create or delete planners through the UI. + +Fixtures used (provided by conftest.py): + caldera_server (session) — base URL string + api_session (session) — authenticated requests.Session + auth_page (function) — Playwright page with auth cookies, not yet navigated + base_url (function) — base URL string + +Run with: + pytest plugins/magma/tests/e2e/test_planners.py -v --browser chromium +""" + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def navigate_to_planners(page: Page, base_url: str) -> None: + """Navigate to the /planners route and wait for network activity to settle.""" + page.goto(f"{base_url}/planners") + page.wait_for_load_state("networkidle") + + +def get_planners_from_api(api_session, base_url: str) -> list: + """Return the list of planner objects from the REST API.""" + resp = api_session.get(f"{base_url}/api/v2/planners") + assert resp.status_code == 200, ( + f"GET /api/v2/planners returned HTTP {resp.status_code}" + ) + return resp.json() + + +# --------------------------------------------------------------------------- +# 1. h2 heading "Planners" is visible +# --------------------------------------------------------------------------- + +def test_planners_page_heading(auth_page: Page, base_url: str) -> None: + """ + Navigating to /planners must render an

    whose text is "Planners". + This confirms the correct Vue view is mounted by the router. + """ + navigate_to_planners(auth_page, base_url) + + heading = auth_page.locator("h2", has_text="Planners") + expect(heading).to_be_visible() + + +# --------------------------------------------------------------------------- +# 2. Introductory description paragraph is visible +# --------------------------------------------------------------------------- + +def test_planners_description_visible(auth_page: Page, base_url: str) -> None: + """ + Below the heading there must be a descriptive paragraph explaining what + a planner is. The text matches the static copy in the Pug template + (begins with "A planner is a module"). + """ + navigate_to_planners(auth_page, base_url) + + # The Pug template renders "A planner is a module that decides which + # abilities a red team agent should execute..." + description = auth_page.locator("p", has_text="A planner is a module") + expect(description).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. GET /api/v2/planners returns a non-empty list +# --------------------------------------------------------------------------- + +def test_planners_api_returns_entries(api_session, base_url: str) -> None: + """ + The planners REST endpoint must return a non-empty JSON array. + Caldera ships with at least one built-in planner (e.g. "sequential"), + so an empty response indicates a configuration problem. + """ + planners = get_planners_from_api(api_session, base_url) + assert len(planners) > 0, ( + "GET /api/v2/planners returned an empty list — " + "at least one built-in planner is expected." + ) + + +# --------------------------------------------------------------------------- +# 4. Planner item count in the UI matches the API count +# --------------------------------------------------------------------------- + +def test_planners_count_matches_ui( + auth_page: Page, base_url: str, api_session +) -> None: + """ + The number of planner cards (or list entries) rendered on the page must + equal the number of objects returned by GET /api/v2/planners. + + Planners are rendered via a v-for loop; each item is expected to be a + card (.card) or a list item (
  • ) inside the main content area. We + try the most specific selector first and fall back gracefully. + """ + planners = get_planners_from_api(api_session, base_url) + api_count = len(planners) + + navigate_to_planners(auth_page, base_url) + + if api_count == 0: + # No planners in the system — nothing should be rendered. + # This path is unexpected for a standard Caldera install. + cards = auth_page.locator(".card") + assert cards.count() == 0, ( + f"API returned 0 planners but {cards.count()} cards are visible" + ) + return + + # Wait for Vue to finish rendering at least one planner entry. + # Planners are typically rendered as .card elements in a v-for loop. + cards = auth_page.locator(".card") + expect(cards.first).to_be_visible() + ui_count = cards.count() + + assert ui_count == api_count, ( + f"UI shows {ui_count} planner card(s) but API returned {api_count}" + ) + + +# --------------------------------------------------------------------------- +# 5. Each planner's name from the API appears on the page +# --------------------------------------------------------------------------- + +def test_planner_names_displayed( + auth_page: Page, base_url: str, api_session +) -> None: + """ + After navigating to /planners, every planner name returned by + GET /api/v2/planners must appear somewhere on the page, confirming + the Vue component renders all API data correctly. + + Skipped when the API returns no planners (unexpected for a live server). + """ + planners = get_planners_from_api(api_session, base_url) + + if not planners: + pytest.skip("No planners available — cannot verify name rendering.") + + navigate_to_planners(auth_page, base_url) + + for planner in planners: + name = planner.get("name", "") + assert name, f"Planner entry is missing a 'name' field: {planner}" + + name_locator = auth_page.locator(f"text={name}") + expect(name_locator.first).to_be_visible( + timeout=5000 + ), f"Planner name '{name}' was not found on the /planners page." + + +# --------------------------------------------------------------------------- +# 6. No "Create" or "New" button is present (planners are read-only) +# --------------------------------------------------------------------------- + +def test_planners_are_read_only(auth_page: Page, base_url: str) -> None: + """ + Planners are built-in modules and cannot be created through the UI. + Neither a "Create" button nor a "New" button must be present on the page. + """ + navigate_to_planners(auth_page, base_url) + + # Assert that no button with "Create" or "New" text exists in the DOM. + # `to_be_hidden()` passes when the element is absent or not visible. + create_btn = auth_page.locator("button", has_text="Create") + expect(create_btn).to_be_hidden() + + new_btn = auth_page.locator("button", has_text="New") + expect(new_btn).to_be_hidden() diff --git a/tests/e2e/magma/test_schedules.py b/tests/e2e/magma/test_schedules.py new file mode 100644 index 000000000..aaa8a6296 --- /dev/null +++ b/tests/e2e/magma/test_schedules.py @@ -0,0 +1,220 @@ +""" +Playwright E2E tests for the Caldera Schedules UI view (/schedules). + +These tests verify the page structure, visibility of key elements, modal +interaction, and API/UI consistency for the SchedulesView component. + +Fixtures used (provided by conftest.py): + caldera_server (session) — base URL string + api_session (session) — authenticated requests.Session + auth_page (function) — Playwright page with auth cookies, not yet navigated + base_url (function) — base URL string + +Run with: + pytest plugins/magma/tests/e2e/test_schedules.py -v --browser chromium +""" + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def navigate_to_schedules(page: Page, base_url: str) -> None: + """Navigate to the /schedules route and wait for network activity to settle.""" + page.goto(f"{base_url}/schedules") + page.wait_for_load_state("networkidle") + + +def get_schedules_from_api(api_session, caldera_server: str) -> list: + """Return the list of schedule objects from the REST API.""" + resp = api_session.get(f"{caldera_server}/api/v2/schedules") + assert resp.status_code == 200, ( + f"GET /api/v2/schedules returned HTTP {resp.status_code}" + ) + return resp.json() + + +# --------------------------------------------------------------------------- +# 1. h2 heading "Schedules" is visible +# --------------------------------------------------------------------------- + +def test_schedules_page_heading(auth_page: Page, base_url: str) -> None: + """ + Navigating to /schedules must render an

    element whose text is + "Schedules", confirming the correct view has been routed to. + """ + navigate_to_schedules(auth_page, base_url) + + heading = auth_page.locator("h2", has_text="Schedules") + expect(heading).to_be_visible() + + +# --------------------------------------------------------------------------- +# 2. Introductory description paragraph is visible +# --------------------------------------------------------------------------- + +def test_schedules_description_visible(auth_page: Page, base_url: str) -> None: + """ + A descriptive paragraph explaining that schedules allow automatic operation + runs must be visible below the heading. The text is part of the static + template copy rendered by the SchedulesView component. + """ + navigate_to_schedules(auth_page, base_url) + + # The paragraph is introduced by the template with "Schedules allow you to + # automatically run an operation at a given time" + description = auth_page.locator("p", has_text="automatically") + expect(description).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. "Create a schedule" primary button is visible +# --------------------------------------------------------------------------- + +def test_create_schedule_button_visible(auth_page: Page, base_url: str) -> None: + """ + The primary call-to-action button labelled "Create a schedule" must be + present and visible in the button toolbar area. + """ + navigate_to_schedules(auth_page, base_url) + + create_btn = auth_page.locator("button", has_text="Create a schedule") + expect(create_btn).to_be_visible() + + +# --------------------------------------------------------------------------- +# 4. Clicking "Create a schedule" opens a modal with form fields +# --------------------------------------------------------------------------- + +def test_create_schedule_modal_opens(auth_page: Page, base_url: str) -> None: + """ + Clicking the "Create a schedule" button must open a modal dialog that + contains at least one input field, indicating the create-schedule form + is active and ready to receive user input. + """ + navigate_to_schedules(auth_page, base_url) + + create_btn = auth_page.locator("button", has_text="Create a schedule") + expect(create_btn).to_be_visible() + create_btn.click() + + # A Bulma modal becomes active by gaining the "is-active" class. + # Wait for the transition before asserting. + modal = auth_page.locator(".modal.is-active") + expect(modal).to_be_visible() + + # The modal must contain at least one input or select element (the form). + form_field = modal.locator("input, select, textarea").first + expect(form_field).to_be_visible() + + +# --------------------------------------------------------------------------- +# 5. UI schedule item count matches API response count +# --------------------------------------------------------------------------- + +def test_schedules_api_count_matches_ui( + auth_page: Page, base_url: str, api_session, caldera_server: str +) -> None: + """ + The number of schedule entries rendered in the schedule list must equal + the number of schedule objects returned by GET /api/v2/schedules. + + When the API returns zero schedules the test verifies that no list items + are present; when schedules exist, the DOM row count must match. + """ + schedules = get_schedules_from_api(api_session, caldera_server) + api_count = len(schedules) + + navigate_to_schedules(auth_page, base_url) + + if api_count == 0: + # With no schedules the list should be empty — verify no schedule rows. + # Schedule entries are expected to be rendered as list items or table + # rows in a repeating block; accept zero as confirmation of empty state. + rows = auth_page.locator(".schedule-item, tbody tr, [data-schedule]") + assert rows.count() == 0, ( + f"API returned 0 schedules but {rows.count()} items are visible in the UI" + ) + return + + # When schedules exist wait for at least one row to appear. + # Schedule entries are rendered inside a repeating list; rows may be

    • + # elements inside a table or generic container items depending on template. + # Try table rows first, then fall back to a broader list-item selector. + rows = auth_page.locator("tbody tr") + if rows.count() == 0: + rows = auth_page.locator(".schedule-item, [data-schedule]") + + expect(rows.first).to_be_visible() + ui_count = rows.count() + + assert ui_count == api_count, ( + f"UI shows {ui_count} schedule row(s) but API returned {api_count}" + ) + + +# --------------------------------------------------------------------------- +# 6. First schedule's name from the API appears on the page +# --------------------------------------------------------------------------- + +def test_schedule_names_displayed( + auth_page: Page, base_url: str, api_session, caldera_server: str +) -> None: + """ + When at least one schedule exists, the name of the first schedule returned + by GET /api/v2/schedules must be visible somewhere on the page, confirming + that the Vue component has rendered data fetched from the API. + + Skipped when no schedules are present in the system. + """ + schedules = get_schedules_from_api(api_session, caldera_server) + + if not schedules: + pytest.skip("No schedules present — cannot verify name display.") + + first_name = schedules[0].get("name", "") + if not first_name: + pytest.skip("First schedule has no name field — skipping.") + + navigate_to_schedules(auth_page, base_url) + + name_locator = auth_page.locator(f"text={first_name}") + expect(name_locator.first).to_be_visible() + + +# --------------------------------------------------------------------------- +# 7. Schedule cron expression is visible on the page +# --------------------------------------------------------------------------- + +def test_schedule_cron_expression_displayed( + auth_page: Page, base_url: str, api_session, caldera_server: str +) -> None: + """ + When at least one schedule has a non-empty 'schedule' (cron) field, that + cron expression string must appear somewhere visible on the /schedules page. + + This confirms the template renders the schedule expression alongside the + schedule name in the list. + + Skipped when no schedules with a cron expression are present. + """ + schedules = get_schedules_from_api(api_session, caldera_server) + + # Find the first schedule that has a non-empty schedule/cron field. + target_cron: str | None = None + for sched in schedules: + cron = sched.get("schedule", "") + if cron: + target_cron = cron + break + + if target_cron is None: + pytest.skip("No schedules with a cron expression found — skipping.") + + navigate_to_schedules(auth_page, base_url) + + cron_locator = auth_page.locator(f"text={target_cron}") + expect(cron_locator.first).to_be_visible() diff --git a/tests/e2e/magma/test_settings.py b/tests/e2e/magma/test_settings.py new file mode 100644 index 000000000..bc69a0ef2 --- /dev/null +++ b/tests/e2e/magma/test_settings.py @@ -0,0 +1,90 @@ +""" +E2E tests for the Caldera Settings page (SettingsView.vue). + +Settings displays Caldera's current main configuration and allows some +values to be updated via a code editor. It also lists active plugins. + +Run with: + pytest plugins/magma/tests/e2e/test_settings.py -v --browser chromium +""" + +from playwright.sync_api import Page, expect + + +def test_settings_page_loads(auth_page: Page, base_url: str) -> None: + """Navigating to /settings stays on /settings (no redirect to /login).""" + auth_page.goto(base_url + '/settings') + auth_page.wait_for_load_state('networkidle') + assert '/login' not in auth_page.url, ( + f'Expected to stay on /settings but ended up at {auth_page.url}' + ) + + +def test_settings_config_api_returns_data(api_session, base_url: str) -> None: + """GET /api/v2/config/main returns a dict with at least a 'port' key.""" + resp = api_session.get(f'{base_url}/api/v2/config/main') + assert resp.status_code == 200 + config = resp.json() + assert isinstance(config, dict) + assert 'port' in config, f"'port' key missing from config: {list(config.keys())}" + + +def test_settings_displays_port_value(auth_page: Page, api_session, base_url: str) -> None: + """The server port from the API config appears somewhere on the settings page.""" + resp = api_session.get(f'{base_url}/api/v2/config/main') + port = str(resp.json().get('port', '')) + assert port, "Could not determine port from config API" + + auth_page.goto(base_url + '/settings') + auth_page.wait_for_load_state('networkidle') + expect(auth_page.get_by_text(port, exact=False)).to_be_visible() + + +def test_settings_plugins_api_returns_list(api_session, base_url: str) -> None: + """GET /api/v2/plugins returns a non-empty list of plugin objects.""" + resp = api_session.get(f'{base_url}/api/v2/plugins') + assert resp.status_code == 200 + plugins = resp.json() + assert isinstance(plugins, list) + assert len(plugins) > 0, "Expected at least one plugin to be registered" + + +def test_settings_page_has_code_editor(auth_page: Page, base_url: str) -> None: + """A code editor element is present on the settings page for config editing.""" + auth_page.goto(base_url + '/settings') + auth_page.wait_for_load_state('networkidle') + # CodeEditor component renders a prism-editor or textarea + editor = auth_page.locator( + '.prism-editor__textarea, textarea, .code-editor, [contenteditable="true"]' + ).first + expect(editor).to_be_visible() + + +def test_settings_api_key_not_in_page_source(auth_page: Page, api_session, base_url: str) -> None: + """ + The raw api_key_red value from the config API should NOT appear verbatim + in the rendered page — it must be masked or omitted in the UI. + """ + config_resp = api_session.get(f'{base_url}/api/v2/config/main') + api_key = config_resp.json().get('api_key_red', '') + + if not api_key or api_key.startswith('$argon2'): + # Hashed key — already opaque, skip this check + return + + auth_page.goto(base_url + '/settings') + auth_page.wait_for_load_state('networkidle') + page_text = auth_page.content() + assert api_key not in page_text, ( + 'Raw api_key_red value is visible in the settings page source — ' + 'it should be masked or omitted for security.' + ) + + +def test_settings_navigation_links_visible(auth_page: Page, base_url: str) -> None: + """Navigation links to other sections are present on the settings page.""" + auth_page.goto(base_url + '/settings') + auth_page.wait_for_load_state('networkidle') + # The sidebar navigation should link to core views + for label in ('Agents', 'Operations', 'Abilities'): + expect(auth_page.get_by_role('link', name=label, exact=False)).to_be_visible() diff --git a/tox.ini b/tox.ini index 1962eec90..91a4ab1b1 100644 --- a/tox.ini +++ b/tox.ini @@ -26,7 +26,7 @@ deps = allowlist_externals = mkdir commands = mkdir -p plugins/magma/dist/assets - coverage run -p -m pytest --tb=short --asyncio-mode=auto tests -vv + coverage run -p -m pytest --tb=short --asyncio-mode=auto --ignore=tests/e2e tests -vv [testenv:style] deps = pre-commit @@ -71,3 +71,31 @@ deps = skip_install = true commands = bandit -r app -ll --exclude=tests/ --skip=B303 + +[testenv:ui] +description = + Run Playwright E2E UI tests against a live Caldera instance. + Each run picks a free port and writes conf/local.yml automatically, + so multiple VENVs can run in parallel without port conflicts. + Requires Node.js to build the Magma frontend first (handled below via nodeenv). +deps = + -rrequirements.txt + -rrequirements-ui-tests.txt + pytest + pytest-asyncio==0.26.0 + pyyaml +allowlist_externals = + bash + nodeenv + npm + playwright +skip_install = true +commands = + # 1. Install Node.js into this virtualenv via nodeenv (idempotent) + nodeenv --node=20.11.0 --python-virtualenv --prebuilt --quiet + # 2. Install Magma frontend dependencies and build the static assets + bash -c "cd plugins/magma && npm ci --prefer-offline && npm run build" + # 3. Install Playwright browser binaries (Chromium only for speed) + playwright install chromium + # 4. Run the E2E test suite + pytest plugins/magma/tests/e2e --tb=short -v {posargs}