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="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="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        )
159
160        self.assertTrue(result)
161        mock_post.assert_called_once()
162
163        # Verify the POST was made with correct data
164        call_kwargs = mock_post.call_args[1]
165        self.assertEqual(call_kwargs["timeout"], 10)
166        self.assertEqual(
167            call_kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded"
168        )
169
170        # Verify form data contains SAMLRequest
171        form_data = call_kwargs["data"]
172        self.assertIn("SAMLRequest", form_data)
173
174    def test_provider_not_found(self):
175        """Test returns False when provider doesn't exist"""
176        result = send_saml_logout_request(
177            provider_pk=99999,  # Non-existent provider
178            sls_url="https://sp.example.com/sls",
179            name_id="test@example.com",
180            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
181            session_index="test-session-123",
182        )
183
184        self.assertFalse(result)
185
186    @patch("authentik.providers.saml.tasks.requests.post")
187    def test_http_error_raises(self, mock_post):
188        """Test HTTP error raises exception (no try/catch in send_post_logout_request)"""
189        mock_response = MagicMock()
190        mock_response.status_code = 500
191        mock_response.raise_for_status.side_effect = HTTPError("500 Server Error")
192        mock_post.return_value = mock_response
193
194        with self.assertRaises(HTTPError):
195            send_saml_logout_request(
196                provider_pk=self.provider.pk,
197                sls_url=self.provider.sls_url,
198                name_id="test@example.com",
199                name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
200                session_index="test-session-123",
201            )
202
203
204class TestSendPostLogoutRequest(TestCase):
205    """Tests for send_post_logout_request function"""
206
207    def setUp(self):
208        """Set up test fixtures"""
209        self.cert = create_test_cert()
210        self.flow = create_test_flow()
211
212        self.provider = SAMLProvider.objects.create(
213            name="test-provider",
214            authorization_flow=self.flow,
215            acs_url="https://sp.example.com/acs",
216            sls_url="https://sp.example.com/sls",
217            issuer="https://idp.example.com",
218            signing_kp=self.cert,
219        )
220
221    @patch("authentik.providers.saml.tasks.requests.post")
222    def test_successful_post(self, mock_post):
223        """Test successful POST returns True"""
224        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
225
226        mock_response = MagicMock()
227        mock_response.status_code = 200
228        mock_response.raise_for_status = MagicMock()
229        mock_post.return_value = mock_response
230
231        processor = LogoutRequestProcessor(
232            provider=self.provider,
233            user=None,
234            destination=self.provider.sls_url,
235            name_id="test@example.com",
236            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
237            session_index="test-session-123",
238        )
239
240        result = send_post_logout_request(self.provider, processor)
241
242        self.assertTrue(result)
243        mock_post.assert_called_once()
244
245    @patch("authentik.providers.saml.tasks.requests.post")
246    def test_with_relay_state(self, mock_post):
247        """Test POST includes RelayState when present"""
248        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
249
250        mock_response = MagicMock()
251        mock_response.status_code = 200
252        mock_response.raise_for_status = MagicMock()
253        mock_post.return_value = mock_response
254
255        processor = LogoutRequestProcessor(
256            provider=self.provider,
257            user=None,
258            destination=self.provider.sls_url,
259            name_id="test@example.com",
260            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
261            session_index="test-session-123",
262            relay_state="https://sp.example.com/return",
263        )
264
265        result = send_post_logout_request(self.provider, processor)
266
267        self.assertTrue(result)
268
269        # Verify RelayState is included
270        form_data = mock_post.call_args[1]["data"]
271        self.assertIn("RelayState", form_data)
272        self.assertEqual(form_data["RelayState"], "https://sp.example.com/return")
273
274    @patch("authentik.providers.saml.tasks.requests.post")
275    def test_connection_error_raises(self, mock_post):
276        """Test connection error raises exception"""
277        from authentik.providers.saml.processors.logout_request import LogoutRequestProcessor
278
279        mock_post.side_effect = ConnectionError("Connection refused")
280
281        processor = LogoutRequestProcessor(
282            provider=self.provider,
283            user=None,
284            destination=self.provider.sls_url,
285            name_id="test@example.com",
286            name_id_format=SAML_NAME_ID_FORMAT_EMAIL,
287            session_index="test-session-123",
288        )
289
290        with self.assertRaises(ConnectionError):
291            send_post_logout_request(self.provider, processor)