authentik.enterprise.endpoints.connectors.google_chrome.tests.test_view

 1from json import dumps
 2from unittest.mock import MagicMock, patch
 3
 4from django.urls import reverse
 5
 6from authentik.core.tests.utils import RequestFactory, create_test_flow
 7from authentik.endpoints.models import Device, EndpointStage
 8from authentik.enterprise.endpoints.connectors.google_chrome.models import GoogleChromeConnector
 9from authentik.enterprise.providers.google_workspace.clients.test_http import MockHTTP
10from authentik.flows.models import FlowStageBinding
11from authentik.flows.planner import PLAN_CONTEXT_DEVICE
12from authentik.flows.tests import FlowTestCase
13from authentik.lib.generators import generate_id
14from authentik.lib.tests.utils import load_fixture
15
16
17class TestChromeDTCView(FlowTestCase):
18    def setUp(self):
19        self.flow = create_test_flow()
20        self.connector = GoogleChromeConnector.objects.create(
21            name=generate_id(),
22            credentials={},
23        )
24        self.factory = RequestFactory()
25        self.api_key = generate_id()
26        self.stage = EndpointStage.objects.create(
27            name=generate_id(),
28            connector=self.connector,
29        )
30        FlowStageBinding.objects.create(
31            target=self.flow,
32            stage=self.stage,
33            order=0,
34        )
35
36    def test_dtc_generate_verify(self):
37        res = self.client.get(
38            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
39        )
40        self.assertStageResponse(
41            res,
42            self.flow,
43            component="xak-flow-frame",
44            url="http://testserver/endpoints/google/chrome/",
45        )
46
47        challenge = generate_id()
48        http = MockHTTP()
49        http.add_response(
50            f"https://verifiedaccess.googleapis.com/v2/challenge:generate?key={self.api_key}&alt=json",
51            {"challenge": challenge},
52            method="POST",
53        )
54        http.add_response(
55            f"https://verifiedaccess.googleapis.com/v2/challenge:verify?key={self.api_key}&alt=json",
56            load_fixture("fixtures/host_macos.json"),
57            method="POST",
58        )
59        with patch(
60            "authentik.enterprise.endpoints.connectors.google_chrome.models.GoogleChromeConnector.google_credentials",
61            MagicMock(return_value={"developerKey": self.api_key, "http": http}),
62        ):
63            # Generate challenge
64            res = self.client.get(
65                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
66                HTTP_X_DEVICE_TRUST="VerifiedAccess",
67            )
68            self.assertEqual(res.status_code, 302)
69            self.assertEqual(
70                res.headers["X-Verified-Access-Challenge"],
71                dumps({"challenge": challenge}),
72            )
73
74            # Validate challenge
75            res = self.client.get(
76                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
77                HTTP_X_VERIFIED_ACCESS_CHALLENGE_RESPONSE=dumps({}),
78            )
79            self.assertEqual(res.status_code, 200)
80        device = Device.objects.get(identifier="Z5DDF07GK6")
81        self.assertIsNotNone(device)
82
83        # Continue flow
84        with self.assertFlowFinishes() as plan:
85            res = self.client.post(
86                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
87            )
88            self.assertStageRedirects(res, "/")
89        plan = plan()
90        plan_device = plan.context[PLAN_CONTEXT_DEVICE]
91        self.assertEqual(device.pk, plan_device.pk)
class TestChromeDTCView(authentik.flows.tests.FlowTestCase):
18class TestChromeDTCView(FlowTestCase):
19    def setUp(self):
20        self.flow = create_test_flow()
21        self.connector = GoogleChromeConnector.objects.create(
22            name=generate_id(),
23            credentials={},
24        )
25        self.factory = RequestFactory()
26        self.api_key = generate_id()
27        self.stage = EndpointStage.objects.create(
28            name=generate_id(),
29            connector=self.connector,
30        )
31        FlowStageBinding.objects.create(
32            target=self.flow,
33            stage=self.stage,
34            order=0,
35        )
36
37    def test_dtc_generate_verify(self):
38        res = self.client.get(
39            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
40        )
41        self.assertStageResponse(
42            res,
43            self.flow,
44            component="xak-flow-frame",
45            url="http://testserver/endpoints/google/chrome/",
46        )
47
48        challenge = generate_id()
49        http = MockHTTP()
50        http.add_response(
51            f"https://verifiedaccess.googleapis.com/v2/challenge:generate?key={self.api_key}&alt=json",
52            {"challenge": challenge},
53            method="POST",
54        )
55        http.add_response(
56            f"https://verifiedaccess.googleapis.com/v2/challenge:verify?key={self.api_key}&alt=json",
57            load_fixture("fixtures/host_macos.json"),
58            method="POST",
59        )
60        with patch(
61            "authentik.enterprise.endpoints.connectors.google_chrome.models.GoogleChromeConnector.google_credentials",
62            MagicMock(return_value={"developerKey": self.api_key, "http": http}),
63        ):
64            # Generate challenge
65            res = self.client.get(
66                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
67                HTTP_X_DEVICE_TRUST="VerifiedAccess",
68            )
69            self.assertEqual(res.status_code, 302)
70            self.assertEqual(
71                res.headers["X-Verified-Access-Challenge"],
72                dumps({"challenge": challenge}),
73            )
74
75            # Validate challenge
76            res = self.client.get(
77                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
78                HTTP_X_VERIFIED_ACCESS_CHALLENGE_RESPONSE=dumps({}),
79            )
80            self.assertEqual(res.status_code, 200)
81        device = Device.objects.get(identifier="Z5DDF07GK6")
82        self.assertIsNotNone(device)
83
84        # Continue flow
85        with self.assertFlowFinishes() as plan:
86            res = self.client.post(
87                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
88            )
89            self.assertStageRedirects(res, "/")
90        plan = plan()
91        plan_device = plan.context[PLAN_CONTEXT_DEVICE]
92        self.assertEqual(device.pk, plan_device.pk)

Helpers for testing flows and stages.

def setUp(self):
19    def setUp(self):
20        self.flow = create_test_flow()
21        self.connector = GoogleChromeConnector.objects.create(
22            name=generate_id(),
23            credentials={},
24        )
25        self.factory = RequestFactory()
26        self.api_key = generate_id()
27        self.stage = EndpointStage.objects.create(
28            name=generate_id(),
29            connector=self.connector,
30        )
31        FlowStageBinding.objects.create(
32            target=self.flow,
33            stage=self.stage,
34            order=0,
35        )

Hook method for setting up the test fixture before exercising it.

def test_dtc_generate_verify(self):
37    def test_dtc_generate_verify(self):
38        res = self.client.get(
39            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
40        )
41        self.assertStageResponse(
42            res,
43            self.flow,
44            component="xak-flow-frame",
45            url="http://testserver/endpoints/google/chrome/",
46        )
47
48        challenge = generate_id()
49        http = MockHTTP()
50        http.add_response(
51            f"https://verifiedaccess.googleapis.com/v2/challenge:generate?key={self.api_key}&alt=json",
52            {"challenge": challenge},
53            method="POST",
54        )
55        http.add_response(
56            f"https://verifiedaccess.googleapis.com/v2/challenge:verify?key={self.api_key}&alt=json",
57            load_fixture("fixtures/host_macos.json"),
58            method="POST",
59        )
60        with patch(
61            "authentik.enterprise.endpoints.connectors.google_chrome.models.GoogleChromeConnector.google_credentials",
62            MagicMock(return_value={"developerKey": self.api_key, "http": http}),
63        ):
64            # Generate challenge
65            res = self.client.get(
66                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
67                HTTP_X_DEVICE_TRUST="VerifiedAccess",
68            )
69            self.assertEqual(res.status_code, 302)
70            self.assertEqual(
71                res.headers["X-Verified-Access-Challenge"],
72                dumps({"challenge": challenge}),
73            )
74
75            # Validate challenge
76            res = self.client.get(
77                reverse("authentik_endpoints_connectors_google_chrome:chrome"),
78                HTTP_X_VERIFIED_ACCESS_CHALLENGE_RESPONSE=dumps({}),
79            )
80            self.assertEqual(res.status_code, 200)
81        device = Device.objects.get(identifier="Z5DDF07GK6")
82        self.assertIsNotNone(device)
83
84        # Continue flow
85        with self.assertFlowFinishes() as plan:
86            res = self.client.post(
87                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
88            )
89            self.assertStageRedirects(res, "/")
90        plan = plan()
91        plan_device = plan.context[PLAN_CONTEXT_DEVICE]
92        self.assertEqual(device.pk, plan_device.pk)