280 lines
9.1 KiB
Python
280 lines
9.1 KiB
Python
"""Tests for downloadable Windows Agent enrollment packages."""
|
|
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.agent_enrollment.router import router
|
|
from app.agent_packages.windows import (
|
|
AgentWindowsPackageError,
|
|
AgentWindowsPackageSourceError,
|
|
WINDOWS_PACKAGE_FILES,
|
|
WINDOWS_PACKAGE_ROOT,
|
|
build_windows_agent_package,
|
|
normalize_hub_url,
|
|
)
|
|
from app.identity import get_current_user, required_api_permission
|
|
|
|
|
|
TOKEN = "t" * 43
|
|
|
|
|
|
class WindowsPackageBuilderTests(unittest.TestCase):
|
|
def source_tree(self, root: Path):
|
|
for name in WINDOWS_PACKAGE_FILES:
|
|
(root / name).write_text(f"source:{name}\n")
|
|
|
|
def test_package_contains_only_generic_source_plus_bootstrap_material(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
self.source_tree(root)
|
|
|
|
package = build_windows_agent_package(
|
|
enrollment_token=TOKEN,
|
|
hub_url="https://minerops.example.test",
|
|
source_root=root,
|
|
)
|
|
|
|
with zipfile.ZipFile(io.BytesIO(package)) as archive:
|
|
names = set(archive.namelist())
|
|
|
|
expected = {
|
|
f"{WINDOWS_PACKAGE_ROOT}/{name}"
|
|
for name in WINDOWS_PACKAGE_FILES
|
|
}
|
|
|
|
expected.update(
|
|
{
|
|
f"{WINDOWS_PACKAGE_ROOT}/enrollment.json",
|
|
f"{WINDOWS_PACKAGE_ROOT}/INSTALL.txt",
|
|
}
|
|
)
|
|
|
|
self.assertEqual(names, expected)
|
|
|
|
enrollment = json.loads(
|
|
archive.read(
|
|
f"{WINDOWS_PACKAGE_ROOT}/enrollment.json"
|
|
)
|
|
)
|
|
|
|
self.assertEqual(enrollment["token"], TOKEN)
|
|
self.assertEqual(enrollment["workflow"], "new_agent")
|
|
self.assertEqual(
|
|
enrollment["hub_url"],
|
|
"https://minerops.example.test",
|
|
)
|
|
|
|
serialized = json.dumps(enrollment)
|
|
|
|
self.assertNotIn("agent_secret", serialized)
|
|
self.assertNotIn("api3_password", serialized)
|
|
self.assertNotIn("api3_account", serialized)
|
|
self.assertNotIn("agent-config.json", "\n".join(names))
|
|
|
|
def test_existing_agent_transition_package_is_generic_and_distinct(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
self.source_tree(root)
|
|
package = build_windows_agent_package(
|
|
enrollment_token=TOKEN,
|
|
hub_url="https://minerops.example.test",
|
|
workflow="existing_agent_credential_transition",
|
|
source_root=root,
|
|
)
|
|
|
|
with zipfile.ZipFile(io.BytesIO(package)) as archive:
|
|
enrollment = json.loads(
|
|
archive.read(
|
|
f"{WINDOWS_PACKAGE_ROOT}/enrollment.json"
|
|
)
|
|
)
|
|
instructions = archive.read(
|
|
f"{WINDOWS_PACKAGE_ROOT}/INSTALL.txt"
|
|
).decode("utf-8")
|
|
names = set(archive.namelist())
|
|
|
|
self.assertEqual(
|
|
enrollment,
|
|
{
|
|
"hub_url": "https://minerops.example.test",
|
|
"token": TOKEN,
|
|
"workflow": "existing_agent_credential_transition",
|
|
},
|
|
)
|
|
self.assertIn("Upgrade-MinerOpsAgentCredential.ps1", instructions)
|
|
self.assertIn(
|
|
f"{WINDOWS_PACKAGE_ROOT}/Upgrade-MinerOpsAgentCredential.ps1",
|
|
names,
|
|
)
|
|
serialized = json.dumps(enrollment)
|
|
for forbidden in (
|
|
"plexus-windows-agent",
|
|
"site_id",
|
|
"agent_zone_id",
|
|
"agent_secret",
|
|
):
|
|
self.assertNotIn(forbidden, serialized)
|
|
|
|
def test_missing_canonical_source_fails_closed(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
self.source_tree(root)
|
|
|
|
(root / WINDOWS_PACKAGE_FILES[0]).unlink()
|
|
|
|
with self.assertRaises(AgentWindowsPackageSourceError):
|
|
build_windows_agent_package(
|
|
enrollment_token=TOKEN,
|
|
hub_url="https://minerops.example.test",
|
|
source_root=root,
|
|
)
|
|
|
|
def test_hub_url_is_origin_only(self):
|
|
self.assertEqual(
|
|
normalize_hub_url("https://minerops.example.test/"),
|
|
"https://minerops.example.test",
|
|
)
|
|
|
|
for invalid in (
|
|
"ftp://minerops.example.test",
|
|
"https://user:pass@minerops.example.test",
|
|
"https://minerops.example.test/path",
|
|
"https://minerops.example.test/?token=x",
|
|
):
|
|
with self.subTest(value=invalid):
|
|
with self.assertRaises(AgentWindowsPackageError):
|
|
normalize_hub_url(invalid)
|
|
|
|
|
|
class WindowsPackageRouteTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = FastAPI()
|
|
self.app.include_router(router)
|
|
self.client = TestClient(self.app)
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
|
|
def authenticate(self, permissions):
|
|
self.app.dependency_overrides[get_current_user] = lambda: {
|
|
"id": "00000000-0000-4000-8000-000000009100",
|
|
"username": "package-test",
|
|
"display_name": "Package Test",
|
|
"role": "Test",
|
|
"permissions": tuple(permissions),
|
|
}
|
|
|
|
def test_permission_mapping_is_manage_only(self):
|
|
self.assertEqual(
|
|
required_api_permission(
|
|
"POST",
|
|
"/api/agent-enrollments/windows-package",
|
|
),
|
|
"agents.manage",
|
|
)
|
|
self.assertEqual(
|
|
required_api_permission(
|
|
"POST",
|
|
"/api/agent-credential-transitions/windows-package",
|
|
),
|
|
"agents.manage",
|
|
)
|
|
|
|
def test_view_only_user_cannot_download_package(self):
|
|
self.authenticate(("agents.view",))
|
|
|
|
response = self.client.post(
|
|
"/api/agent-enrollments/windows-package",
|
|
headers={"Origin": "http://testserver"},
|
|
json={
|
|
"token": TOKEN,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
def test_manager_download_is_same_origin_and_not_cacheable(self):
|
|
self.authenticate(("agents.view", "agents.manage"))
|
|
|
|
response = self.client.post(
|
|
"/api/agent-enrollments/windows-package",
|
|
headers={"Origin": "http://testserver"},
|
|
json={
|
|
"token": TOKEN,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(
|
|
response.headers["content-type"],
|
|
"application/zip",
|
|
)
|
|
self.assertEqual(
|
|
response.headers["cache-control"],
|
|
"no-store",
|
|
)
|
|
self.assertIn(
|
|
'filename="MinerOpsAgent-Windows.zip"',
|
|
response.headers["content-disposition"],
|
|
)
|
|
|
|
with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
|
|
enrollment = json.loads(
|
|
archive.read(
|
|
f"{WINDOWS_PACKAGE_ROOT}/enrollment.json"
|
|
)
|
|
)
|
|
self.assertEqual(enrollment["token"], TOKEN)
|
|
self.assertEqual(enrollment["workflow"], "new_agent")
|
|
|
|
def test_manager_downloads_distinct_existing_agent_upgrade_package(self):
|
|
self.authenticate(("agents.view", "agents.manage"))
|
|
|
|
response = self.client.post(
|
|
"/api/agent-credential-transitions/windows-package",
|
|
headers={"Origin": "http://testserver"},
|
|
json={"token": TOKEN},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.headers["cache-control"], "no-store")
|
|
self.assertIn(
|
|
"MinerOpsAgent-Credential-Upgrade-Windows.zip",
|
|
response.headers["content-disposition"],
|
|
)
|
|
with zipfile.ZipFile(io.BytesIO(response.content)) as archive:
|
|
enrollment = json.loads(
|
|
archive.read(f"{WINDOWS_PACKAGE_ROOT}/enrollment.json")
|
|
)
|
|
instructions = archive.read(
|
|
f"{WINDOWS_PACKAGE_ROOT}/INSTALL.txt"
|
|
).decode("utf-8")
|
|
self.assertEqual(
|
|
enrollment["workflow"],
|
|
"existing_agent_credential_transition",
|
|
)
|
|
self.assertIn("Upgrade-MinerOpsAgentCredential.ps1", instructions)
|
|
|
|
def test_package_requires_browser_origin(self):
|
|
self.authenticate(("agents.view", "agents.manage"))
|
|
|
|
response = self.client.post(
|
|
"/api/agent-enrollments/windows-package",
|
|
json={
|
|
"token": TOKEN,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|