authentik.providers.saml.tests.test_tasks

  1"""Tests for SAML provider tasks"""
  2
  3from unittest.mock import MagicMock, patch
  4
  5from django.test import TestCase
  6from requests.exceptions import ConnectionError, HTTPError
  7
  8from authentik.common.saml.constants import SAML_NAME_ID_FORMAT_EMAIL
  9from authentik.core.tests.utils import create_test_cert, create_test_flow
 10from authentik.providers.saml.models import SAMLProvider
 11from authentik.providers.saml.tasks import (
 12    send_post_logout_request,
 13    send_saml_logout_request,
 14    send_saml_logout_response,
 15)
 16
 17
 18class TestSendSamlLogoutResponse(TestCase):
 19    """Tests for send_saml_logout_response task"""
 20
 21    def setUp(self):
 22        """Set up test fixtures"""
 23        self.cert = create_test_cert()
 24        self.flow = create_test_flow()
 25
 26        self.provider = SAMLProvider.objects.create(
 27            name="test-provider",
 28            authorization_flow=self.flow,
 29            acs_url="https://sp.example.com/acs",
 30            sls_url="https://sp.example.com/sls",
 31            issuer_override="https://idp.example.com",
 32            signing_kp=self.cert,
 33        )
 34
 35    @patch("authentik.providers.saml.tasks.requests.post")
 36    def test_successful_logout_response(self, mock_post):
 37        """Test successful POST to SP returns True"""
 38        mock_response = MagicMock()
 39        mock_response.status_code = 200
 40        mock_response.raise_for_status = MagicMock()
 41        mock_post.return_value = mock_response
 42
 43        result = send_saml_logout_response(
 44            provider_pk=self.provider.pk,
 45            sls_url=self.provider.sls_url,
 46            logout_request_id="test-request-id",
 47            relay_state="https://sp.example.com/return",
 48        )
 49
 50        self.assertTrue(result)
 51        mock_post.assert_called_once()
 52
 53        # Verify the POST was made with correct data
 54        call_kwargs = mock_post.call_args[1]
 55        self.assertEqual(call_kwargs["timeout"], 10)
 56        self.assertEqual(
 57            call_kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded"
 58        )
 59
 60        # Verify form data contains SAMLResponse and RelayState
 61        form_data = call_kwargs["data"]
 62        self.assertIn("SAMLResponse", form_data)
 63        self.assertIn("RelayState", form_data)
 64        self.assertEqual(form_data["RelayState"], "https://sp.example.com/return")
 65
 66    @patch("authentik.providers.saml.tasks.requests.post")
 67    def test_successful_logout_response_no_relay_state(self, mock_post):
 68        """Test successful POST without relay_state"""
 69        mock_response = MagicMock()
 70        mock_response.status_code = 200
 71        mock_response.raise_for_status = MagicMock()
 72        mock_post.return_value = mock_response
 73
 74        result = send_saml_logout_response(
 75            provider_pk=self.provider.pk,
 76            sls_url=self.provider.sls_url,
 77            logout_request_id="test-request-id",
 78            relay_state=None,
 79        )
 80
 81        self.assertTrue(result)
 82
 83        # Verify form data does not contain RelayState
 84        form_data = mock_post.call_args[1]["data"]
 85        self.assertIn("SAMLResponse", form_data)
 86        self.assertNotIn("RelayState", form_data)
 87
 88    def test_provider_not_found(self):
 89        """Test returns False when provider doesn't exist"""
 90        result = send_saml_logout_response(
 91            provider_pk=99999,  # Non-existent provider
 92            sls_url="https://sp.example.com/sls",
 93            logout_request_id="test-request-id",
 94            relay_state=None,
 95        )
 96
 97        self.assertFalse(result)
 98
 99    @patch("authentik.providers.saml.tasks.Event")
100    @patch("authentik.providers.saml.tasks.requests.post")
101    def test_http_error_creates_event(self, mock_post, mock_event_class):
102        """Test HTTP error creates an error event"""
103        mock_response = MagicMock()
104        mock_response.status_code = 500
105        mock_response.raise_for_status.side_effect = HTTPError("500 Server Error")
106        mock_post.return_value = mock_response
107
108        mock_event = MagicMock()
109        mock_event_class.new.return_value = mock_event
110
111        result = send_saml_logout_response(
112            provider_pk=self.provider.pk,
113            sls_url=self.provider.sls_url,
114            logout_request_id="test-request-id",
115            relay_state=None,
116        )
117
118        self.assertFalse(result)
119
120        # Verify error event was created
121        mock_event_class.new.assert_called_once()
122        call_kwargs = mock_event_class.new.call_args[1]
123        self.assertIn("Backchannel logout response failed", call_kwargs["message"])
124        mock_event.save.assert_called_once()
125
126
127class TestSendSamlLogoutRequest(TestCase):
128    """Tests for send_saml_logout_request task"""
129
130    def setUp(self):
131        """Set up test fixtures"""
132        self.cert = create_test_cert()
133        self.flow = create_test_flow()
134
135        self.provider = SAMLProvider.objects.create(
136            name="test-provider",
137            authorization_flow=self.flow,
138            acs_url="https://sp.example.com/acs",
139            sls_url="https://sp.example.com/sls",
140            issuer_override="https://idp.example.com",
141            signing_kp=self.cert,
142        )
143
144    @patch("authentik.providers.saml.tasks.requests.post")
145    def test_successful_logout_request(self, mock_post):
146        """Test successful POST logout request returns True"""
147        mock_response = MagicMock()
148        mock_response.status_code = 200
149        mock_response.raise_for_status = MagicMock()
150        mock_post.return_value = mock_response
151
152        result = send_saml_logout_request(
153            provider_pk=self.provider.pk,
154            sls_url=self.provider.sls_url,
155            name_id="test@example.com",
156            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
157            session_index="test-session-123",
158            issuer="https://idp.example.com",
159        )
160
161        self.assertTrue(result)
162        mock_post.assert_called_once()
163
164        # Verify the POST was made with correct data
165        call_kwargs = mock_post.call_args[1]
166        self.assertEqual(call_kwargs["timeout"], 10)
167        self.assertEqual(
168            call_kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded"
169        )
170
171        # Verify form data contains SAMLRequest
172        form_data = call_kwargs["data"]
173        self.assertIn("SAMLRequest", form_data)
174
175    def test_provider_not_found(self):
176        """Test returns False when provider doesn't exist"""
177        result = send_saml_logout_request(
178            provider_pk=99999,  # Non-existent provider
179            sls_url="https://sp.example.com/sls",
180            name_id="test@example.com",
181            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
182            session_index="test-session-123",
183            issuer="https://idp.example.com",
184        )
185
186        self.assertFalse(result)
187
188    @patch("authentik.providers.saml.tasks.requests.post")
189    def test_http_error_raises(self, mock_post):
190        """Test HTTP error raises exception (no try/catch in send_post_logout_request)"""
191        mock_response = MagicMock()
192        mock_response.status_code = 500
193        mock_response.raise_for_status.side_effect = HTTPError("500 Server Error")
194        mock_post.return_value = mock_response
195
196        with self.assertRaises(HTTPError):
197            send_saml_logout_request(
198                provider_pk=self.provider.pk,
199                sls_url=self.provider.sls_url,
200                name_id="test@example.com",
201                name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
202                session_index="test-session-123",
203                issuer="https://idp.example.com",
204            )
205
206
207class TestSendPostLogoutRequest(TestCase):
208    """Tests for send_post_logout_request function"""
209
210    def setUp(self):
211        """Set up test fixtures"""
212        self.cert = create_test_cert()
213        self.flow = create_test_flow()
214
215        self.provider = SAMLProvider.objects.create(
216            name="test-provider",
217            authorization_flow=self.flow,
218            acs_url="https://sp.example.com/acs",
219            sls_url="https://sp.example.com/sls",
220            issuer_override="https://idp.example.com",
221            signing_kp=self.cert,
222        )
223
224    @patch("authentik.providers.saml.tasks.requests.post")
225    def test_successful_post(self, mock_post):
226        """Test successful POST returns True"""
227        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
228
229        mock_response = MagicMock()
230        mock_response.status_code = 200
231        mock_response.raise_for_status = MagicMock()
232        mock_post.return_value = mock_response
233
234        processor = LogoutRequestProcessor(
235            provider=self.provider,
236            user=None,
237            destination=self.provider.sls_url,
238            name_id="test@example.com",
239            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
240            session_index="test-session-123",
241        )
242
243        result = send_post_logout_request(self.provider, processor)
244
245        self.assertTrue(result)
246        mock_post.assert_called_once()
247
248    @patch("authentik.providers.saml.tasks.requests.post")
249    def test_with_relay_state(self, mock_post):
250        """Test POST includes RelayState when present"""
251        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
252
253        mock_response = MagicMock()
254        mock_response.status_code = 200
255        mock_response.raise_for_status = MagicMock()
256        mock_post.return_value = mock_response
257
258        processor = LogoutRequestProcessor(
259            provider=self.provider,
260            user=None,
261            destination=self.provider.sls_url,
262            name_id="test@example.com",
263            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
264            session_index="test-session-123",
265            relay_state="https://sp.example.com/return",
266        )
267
268        result = send_post_logout_request(self.provider, processor)
269
270        self.assertTrue(result)
271
272        # Verify RelayState is included
273        form_data = mock_post.call_args[1]["data"]
274        self.assertIn("RelayState", form_data)
275        self.assertEqual(form_data["RelayState"], "https://sp.example.com/return")
276
277    @patch("authentik.providers.saml.tasks.requests.post")
278    def test_connection_error_raises(self, mock_post):
279        """Test connection error raises exception"""
280        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
281
282        mock_post.side_effect = ConnectionError("Connection refused")
283
284        processor = LogoutRequestProcessor(
285            provider=self.provider,
286            user=None,
287            destination=self.provider.sls_url,
288            name_id="test@example.com",
289            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
290            session_index="test-session-123",
291        )
292
293        with self.assertRaises(ConnectionError):
294            send_post_logout_request(self.provider, processor)