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(),
 25            url="http://localhost",
 26            token=generate_id(),
 27            map_teams_access_group=True,
 28        )
 29
 30    def test_sync(self):
 31        controller = self.connector.controller(self.connector)
 32        with Mocker() as mock:
 33            mock.get(
 34                "http://localhost/api/v1/fleet/conditional_access/idp/apple/profile",
 35                text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
 36            )
 37            mock.get(
 38                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 39                json=TEST_HOST,
 40            )
 41            mock.get(
 42                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 43                json={"hosts": []},
 44            )
 45            controller.sync_endpoints()
 46        device = Device.objects.filter(
 47            identifier="VMware-56 4d 4a 5a b0 22 7b d7-9b a5 0b dc 8f f2 3b 60"
 48        ).first()
 49        self.assertIsNotNone(device)
 50        group = device.access_group
 51        self.assertIsNotNone(group)
 52        self.assertEqual(group.name, "prod")
 53        self.assertEqual(
 54            device.cached_facts.data,
 55            {
 56                "os": {
 57                    "arch": "x86_64",
 58                    "name": "Ubuntu",
 59                    "family": "linux",
 60                    "version": "24.04.3 LTS",
 61                },
 62                "disks": [],
 63                "vendor": {
 64                    "fleetdm.com": {
 65                        "policies": [],
 66                        "agent_version": "",
 67                        "uuid": "5a4a4d56-22b0-d77b-9ba5-0bdc8ff23b60",
 68                    }
 69                },
 70                "network": {"hostname": "ubuntu-desktop", "interfaces": []},
 71                "hardware": {
 72                    "model": "VMware20,1",
 73                    "serial": "VMware-56 4d 4a 5a b0 22 7b d7-9b a5 0b dc 8f f2 3b 60",
 74                    "cpu_count": 2,
 75                    "cpu_name": "Intel(R) Core(TM) i5-10500T CPU @ 2.30GHz",
 76                    "manufacturer": "VMware, Inc.",
 77                    "memory_bytes": 2062721024,
 78                },
 79                "software": [],
 80            },
 81        )
 82
 83    def test_sync_headers(self):
 84        mapping = NotificationWebhookMapping.objects.create(
 85            name=generate_id(), expression="""return {"foo": "bar"}"""
 86        )
 87        self.connector.headers_mapping = mapping
 88        self.connector.save()
 89        controller = self.connector.controller(self.connector)
 90        with Mocker() as mock:
 91            mock.get(
 92                "http://localhost/api/v1/fleet/conditional_access/idp/apple/profile",
 93                text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
 94            )
 95            mock.get(
 96                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
 97                json=TEST_HOST,
 98            )
 99            mock.get(
100                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
101                json={"hosts": []},
102            )
103            controller.sync_endpoints()
104        self.assertEqual(mock.call_count, 3)
105        self.assertEqual(mock.request_history[0].method, "GET")
106        self.assertEqual(mock.request_history[0].headers["foo"], "bar")
107        self.assertEqual(mock.request_history[1].method, "GET")
108        self.assertEqual(mock.request_history[1].headers["foo"], "bar")
109        self.assertEqual(mock.request_history[2].method, "GET")
110        self.assertEqual(mock.request_history[2].headers["foo"], "bar")
111
112    def test_map_host_linux(self):
113        controller = self.connector.controller(self.connector)
114        self.assertEqual(
115            controller.map_os(TEST_HOST_UBUNTU),
116            {
117                "arch": "x86_64",
118                "family": OSFamily.linux,
119                "name": "Ubuntu",
120                "version": "24.04.3 LTS",
121            },
122        )
123        self.assertEqual(
124            controller.map_os(TEST_HOST_FEDORA),
125            {
126                "arch": "x86_64",
127                "family": OSFamily.linux,
128                "name": "Fedora Linux",
129                "version": "43.0.0",
130            },
131        )
132
133    def test_map_host_windows(self):
134        controller = self.connector.controller(self.connector)
135        self.assertEqual(
136            controller.map_os(TEST_HOST_WINDOWS),
137            {
138                "arch": "x86_64",
139                "family": OSFamily.windows,
140                "name": "Windows Server 2022 Datacenter 21H2",
141                "version": "10.0.20348.4405",
142            },
143        )
144
145    def test_map_host_macos(self):
146        controller = self.connector.controller(self.connector)
147        self.assertEqual(
148            controller.map_os(TEST_HOST_MACOS),
149            {
150                "arch": "arm64e",
151                "family": OSFamily.macOS,
152                "name": "macOS",
153                "version": "26.3",
154            },
155        )