authentik.enterprise.endpoints.connectors.fleet.tests.test_connector

  1from json import loads
  2
  3from requests_mock import Mocker
  4from rest_framework.test import APITestCase
  5
  6from authentik.endpoints.facts import OSFamily
  7from authentik.endpoints.models import Device
  8from authentik.enterprise.endpoints.connectors.fleet.models import FleetConnector
  9from authentik.events.models import NotificationWebhookMapping
 10from authentik.lib.generators import generate_id
 11from authentik.lib.tests.utils import load_fixture
 12
 13TEST_HOST_UBUNTU = loads(load_fixture("fixtures/host_ubuntu.json"))
 14TEST_HOST_FEDORA = loads(load_fixture("fixtures/host_fedora.json"))
 15TEST_HOST_MACOS = loads(load_fixture("fixtures/host_macos.json"))
 16TEST_HOST_WINDOWS = loads(load_fixture("fixtures/host_windows.json"))
 17
 18TEST_HOST = {"hosts": [TEST_HOST_UBUNTU, TEST_HOST_MACOS, TEST_HOST_WINDOWS, TEST_HOST_FEDORA]}
 19
 20
 21class TestFleetConnector(APITestCase):
 22    def setUp(self):
 23        self.connector = FleetConnector.objects.create(
 24            name=generate_id(), url="http://localhost", token=generate_id()
 25        )
 26
 27    def test_sync(self):
 28        controller = self.connector.controller(self.connector)
 29        with Mocker() as mock:
 30            mock.get(
 31                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 32                json=TEST_HOST,
 33            )
 34            mock.get(
 35                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 36                json={"hosts": []},
 37            )
 38            controller.sync_endpoints()
 39        device = Device.objects.filter(
 40            identifier="VMware-56 4d 4a 5a b0 22 7b d7-9b a5 0b dc 8f f2 3b 60"
 41        ).first()
 42        self.assertIsNotNone(device)
 43        self.assertEqual(
 44            device.cached_facts.data,
 45            {
 46                "os": {
 47                    "arch": "x86_64",
 48                    "name": "Ubuntu",
 49                    "family": "linux",
 50                    "version": "24.04.3 LTS",
 51                },
 52                "disks": [],
 53                "vendor": {"fleetdm.com": {"policies": [], "agent_version": ""}},
 54                "network": {"hostname": "ubuntu-desktop", "interfaces": []},
 55                "hardware": {
 56                    "model": "VMware20,1",
 57                    "serial": "VMware-56 4d 4a 5a b0 22 7b d7-9b a5 0b dc 8f f2 3b 60",
 58                    "cpu_count": 2,
 59                    "cpu_name": "Intel(R) Core(TM) i5-10500T CPU @ 2.30GHz",
 60                    "manufacturer": "VMware, Inc.",
 61                    "memory_bytes": 2062721024,
 62                },
 63                "software": [],
 64            },
 65        )
 66
 67    def test_sync_headers(self):
 68        mapping = NotificationWebhookMapping.objects.create(
 69            name=generate_id(), expression="""return {"foo": "bar"}"""
 70        )
 71        self.connector.headers_mapping = mapping
 72        self.connector.save()
 73        controller = self.connector.controller(self.connector)
 74        with Mocker() as mock:
 75            mock.get(
 76                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 77                json=TEST_HOST,
 78            )
 79            mock.get(
 80                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 81                json={"hosts": []},
 82            )
 83            controller.sync_endpoints()
 84        self.assertEqual(mock.call_count, 2)
 85        self.assertEqual(mock.request_history[0].method, "GET")
 86        self.assertEqual(mock.request_history[0].headers["foo"], "bar")
 87        self.assertEqual(mock.request_history[1].method, "GET")
 88        self.assertEqual(mock.request_history[1].headers["foo"], "bar")
 89
 90    def test_map_host_linux(self):
 91        controller = self.connector.controller(self.connector)
 92        self.assertEqual(
 93            controller.map_os(TEST_HOST_UBUNTU),
 94            {
 95                "arch": "x86_64",
 96                "family": OSFamily.linux,
 97                "name": "Ubuntu",
 98                "version": "24.04.3 LTS",
 99            },
100        )
101        self.assertEqual(
102            controller.map_os(TEST_HOST_FEDORA),
103            {
104                "arch": "x86_64",
105                "family": OSFamily.linux,
106                "name": "Fedora Linux",
107                "version": "43.0.0",
108            },
109        )
110
111    def test_map_host_windows(self):
112        controller = self.connector.controller(self.connector)
113        self.assertEqual(
114            controller.map_os(TEST_HOST_WINDOWS),
115            {
116                "arch": "x86_64",
117                "family": OSFamily.windows,
118                "name": "Windows Server 2022 Datacenter 21H2",
119                "version": "10.0.20348.4405",
120            },
121        )
122
123    def test_map_host_macos(self):
124        controller = self.connector.controller(self.connector)
125        self.assertEqual(
126            controller.map_os(TEST_HOST_MACOS),
127            {
128                "arch": "arm64e",
129                "family": OSFamily.macOS,
130                "name": "macOS",
131                "version": "26.0.1",
132            },
133        )