feature/m4-mitre #1

Merged
knacky merged 13 commits from feature/m4-mitre into main 2026-05-12 17:24:14 +00:00
2 changed files with 80 additions and 10 deletions
Showing only changes of commit 8b1de6e258 - Show all commits

View File

@@ -8,7 +8,6 @@ from __future__ import annotations
import json import json
import secrets import secrets
import uuid
from pathlib import Path from pathlib import Path
import pytest import pytest
@@ -245,12 +244,17 @@ def test_seed_persists_setting(app, fixture_bundle_path):
assert status["version"] is None # only set when source == MITRE_DEFAULT_URL assert status["version"] is None # only set when source == MITRE_DEFAULT_URL
def test_checksum_mismatch_aborts(tmp_path): def test_checksum_mismatch_aborts(tmp_path, monkeypatch):
"""A wrong sha256 triggers MitreChecksumMismatch and skips DB writes.""" """A wrong sha256 triggers MitreChecksumMismatch and skips DB writes.
We monkey-patch the allowlist to accept `file://` for the duration of the
test — file:// is rejected in production by `_ensure_host_allowed` (cf.
`test_seed_refuses_file_url`), but we need to drive `_download` past that
gate to exercise the sha256 path.
"""
path = tmp_path / "tiny.json" path = tmp_path / "tiny.json"
path.write_text(json.dumps(MINIMAL_BUNDLE)) path.write_text(json.dumps(MINIMAL_BUNDLE))
# Force the URL path so download() is invoked. We mock by passing a file:// URL. monkeypatch.setattr(mitre_svc, "_ensure_host_allowed", lambda _: None)
# Simpler: call _download() directly with a bogus hash.
bogus = "0" * 64 bogus = "0" * 64
with pytest.raises(mitre_svc.MitreChecksumMismatch): with pytest.raises(mitre_svc.MitreChecksumMismatch):
mitre_svc._download( mitre_svc._download(
@@ -387,3 +391,63 @@ def test_matrix_endpoint_requires_auth(app, fixture_bundle_path):
mitre_svc.seed_mitre(source=fixture_bundle_path, expected_sha256=None) mitre_svc.seed_mitre(source=fixture_bundle_path, expected_sha256=None)
with app.test_client() as c: with app.test_client() as c:
assert c.get("/api/v1/mitre/matrix").status_code == 401 assert c.get("/api/v1/mitre/matrix").status_code == 401
# === Security guards ==========================================================
def test_seed_refuses_file_url(tmp_path):
"""file:// (or any scheme outside the allowlist) is rejected — protects
against a privileged operator pivoting the in-container fetch to local
filesystem reads via the URL path."""
path = tmp_path / "bundle.json"
path.write_text(json.dumps(MINIMAL_BUNDLE))
with pytest.raises(mitre_svc.MitreSourceForbidden):
mitre_svc._download(f"file://{path}", tmp_path / "out.json")
def test_seed_refuses_disallowed_https_host(tmp_path):
"""An HTTPS URL outside MITRE_ALLOWED_HOSTS is rejected before any I/O.
Closes the SSRF surface (cloud metadata, internal mirrors)."""
with pytest.raises(mitre_svc.MitreSourceForbidden):
mitre_svc._download("https://attacker.example/bundle.json", tmp_path / "out.json")
def test_seed_refuses_custom_url_without_sha(tmp_path):
"""End-to-end refusal: even an allowlisted custom URL needs a sha or an
explicit allow_unverified=True."""
# Use the default URL with a different sha to simulate "custom" semantics
# without actually hitting the network: pass a different MITRE_DEFAULT_URL.
# The cleanest expression is to call seed_mitre with the same URL but no sha
# — but the default URL gets the default sha auto-set; we need to bypass.
with pytest.raises(mitre_svc.MitreSeedError):
mitre_svc.seed_mitre(
source="https://raw.githubusercontent.com/some-other-path/bundle.json",
expected_sha256=None,
allow_unverified=False,
)
def test_dotted_id_fallback_resolves_orphan_subtechnique(app, tmp_path):
"""When the STIX `subtechnique-of` relationship is missing, the parser
must fall back to the dotted convention (T1003.001 → T1003)."""
bundle = json.loads(json.dumps(MINIMAL_BUNDLE)) # deep copy
# Strip the relationship object so the parent_stix_id lookup fails.
bundle["objects"] = [o for o in bundle["objects"] if o.get("type") != "relationship"]
bundle_path = tmp_path / "no-rel.json"
bundle_path.write_text(json.dumps(bundle))
result = mitre_svc.seed_mitre(source=bundle_path, expected_sha256=None)
# The fallback resolves T1059.001 → T1059 via the dotted-id pattern,
# so the subtechnique is still attached (no orphan).
assert result.subtechniques_upserted == 1
assert result.subtechniques_skipped_orphan == 0
def test_seed_clears_version_when_source_is_not_default(app, fixture_bundle_path):
"""A custom source must NULL `mitre_version` so /mitre/status doesn't lie
about a stale upstream pin."""
# First seed from the default URL would set version=19.0; here we seed from
# a local file path, which should write version=None.
mitre_svc.seed_mitre(source=fixture_bundle_path, expected_sha256=None)
assert mitre_svc.read_status()["version"] is None

View File

@@ -9,7 +9,9 @@ import { expect, test, type APIRequestContext, type Page } from '@playwright/tes
* + the picker UI. * + the picker UI.
*/ */
const ADMIN_EMAIL = `admin-${Math.floor(Math.random() * 1e6)}@metamorph.local`; // crypto.randomUUID() guarantees uniqueness across parallel test runs; the
// Math.random() previous pattern could collide one-in-a-million in CI.
const ADMIN_EMAIL = `admin-${crypto.randomUUID().slice(0, 8)}@metamorph.local`;
const ADMIN_PASSWORD = 'AdminPass1234!'; const ADMIN_PASSWORD = 'AdminPass1234!';
async function resetAndMintToken(request: APIRequestContext): Promise<string> { async function resetAndMintToken(request: APIRequestContext): Promise<string> {
@@ -55,9 +57,13 @@ test.describe('M4 — MITRE ATT&CK reference', () => {
}); });
expect(sync.status(), `mitre sync failed: ${await sync.text()}`).toBe(200); expect(sync.status(), `mitre sync failed: ${await sync.text()}`).toBe(200);
const result = await sync.json(); const result = await sync.json();
expect(result.tactics_upserted).toBeGreaterThanOrEqual(14); // Pinned exactly to MITRE Enterprise v19.0 — bump alongside MITRE_VERSION
expect(result.techniques_upserted).toBeGreaterThanOrEqual(180); // in `app/services/mitre_seed.py` when the pin changes. Exact counts catch
expect(result.subtechniques_upserted).toBeGreaterThanOrEqual(400); // parser regressions that would silently include revoked/deprecated rows.
expect(result.tactics_upserted).toBe(15);
expect(result.techniques_upserted).toBe(222);
expect(result.subtechniques_upserted).toBe(475);
expect(result.subtechniques_skipped_orphan).toBe(0);
}); });
test('GET /mitre/tactics returns 14+ Enterprise tactics', async ({ request }) => { test('GET /mitre/tactics returns 14+ Enterprise tactics', async ({ request }) => {
@@ -146,7 +152,7 @@ test.describe('M4 — MITRE ATT&CK reference', () => {
test('Non-admin cannot trigger /mitre/sync', async ({ page, request }) => { test('Non-admin cannot trigger /mitre/sync', async ({ page, request }) => {
// Invite a no-perm user via the admin. // Invite a no-perm user via the admin.
const adminAccess = await loginAndGetAccess(request, ADMIN_EMAIL, ADMIN_PASSWORD); const adminAccess = await loginAndGetAccess(request, ADMIN_EMAIL, ADMIN_PASSWORD);
const eveEmail = `eve-${Math.floor(Math.random() * 1e6)}@metamorph.local`; const eveEmail = `eve-${crypto.randomUUID().slice(0, 8)}@metamorph.local`;
const inv = await request.post('/api/v1/invitations', { const inv = await request.post('/api/v1/invitations', {
headers: { Authorization: `Bearer ${adminAccess}` }, headers: { Authorization: `Bearer ${adminAccess}` },
data: { email_hint: eveEmail }, data: { email_hint: eveEmail },