From 24e8915e0a40098dd8efe2b3b420a37f2f7d31ec Mon Sep 17 00:00:00 2001 From: Jens L Date: Wed, 4 Jan 2023 22:04:16 +0100 Subject: [PATCH] providers/proxy: add tests for proxy basic auth (#4357) * add tests for proxy basic auth Signed-off-by: Jens Langhammer * stop bandit from complaining Signed-off-by: Jens Langhammer * add API tests Signed-off-by: Jens Langhammer * more tests Signed-off-by: Jens Langhammer Signed-off-by: Jens Langhammer --- authentik/providers/proxy/api.py | 19 ++++- authentik/providers/proxy/tests.py | 122 +++++++++++++++++++++++++++++ tests/e2e/test_provider_ldap.py | 1 - tests/e2e/test_provider_proxy.py | 74 +++++++++++++++++ 4 files changed, 212 insertions(+), 4 deletions(-) create mode 100644 authentik/providers/proxy/tests.py diff --git a/authentik/providers/proxy/api.py b/authentik/providers/proxy/api.py index 2561b4377..0054000f5 100644 --- a/authentik/providers/proxy/api.py +++ b/authentik/providers/proxy/api.py @@ -1,6 +1,7 @@ """ProxyProvider API Views""" from typing import Any, Optional +from django.utils.translation import gettext_lazy as _ from drf_spectacular.utils import extend_schema_field from rest_framework.exceptions import ValidationError from rest_framework.fields import CharField, ListField, ReadOnlyField, SerializerMethodField @@ -39,22 +40,34 @@ class ProxyProviderSerializer(ProviderSerializer): redirect_uris = CharField(read_only=True) outpost_set = ListField(child=CharField(), read_only=True, source="outpost_set.all") + def validate_basic_auth_enabled(self, value: bool) -> bool: + """Ensure user and password attributes are set""" + if value: + if ( + self.initial_data.get("basic_auth_password_attribute", "") == "" + or self.initial_data.get("basic_auth_user_attribute", "") == "" + ): + raise ValidationError( + _("User and password attributes must be set when basic auth is enabled.") + ) + return value + def validate(self, attrs) -> dict[Any, str]: """Check that internal_host is set when mode is Proxy""" if ( attrs.get("mode", ProxyMode.PROXY) == ProxyMode.PROXY and attrs.get("internal_host", "") == "" ): - raise ValidationError("Internal host cannot be empty when forward auth is disabled.") + raise ValidationError(_("Internal host cannot be empty when forward auth is disabled.")) return attrs - def create(self, validated_data): + def create(self, validated_data: dict): instance: ProxyProvider = super().create(validated_data) instance.set_oauth_defaults() instance.save() return instance - def update(self, instance: ProxyProvider, validated_data): + def update(self, instance: ProxyProvider, validated_data: dict): instance = super().update(instance, validated_data) instance.set_oauth_defaults() instance.save() diff --git a/authentik/providers/proxy/tests.py b/authentik/providers/proxy/tests.py new file mode 100644 index 000000000..76409df89 --- /dev/null +++ b/authentik/providers/proxy/tests.py @@ -0,0 +1,122 @@ +"""proxy provider tests""" +from django.urls import reverse +from rest_framework.test import APITestCase + +from authentik.core.tests.utils import create_test_admin_user, create_test_flow +from authentik.lib.generators import generate_id +from authentik.providers.oauth2.models import ClientTypes +from authentik.providers.proxy.models import ProxyMode, ProxyProvider + + +class ProxyProviderTests(APITestCase): + """proxy provider tests""" + + def setUp(self) -> None: + self.user = create_test_admin_user() + self.client.force_login(self.user) + + def test_basic_auth(self): + """Test basic_auth_enabled""" + response = self.client.post( + reverse("authentik_api:proxyprovider-list"), + { + "name": generate_id(), + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + "internal_host": "http://localhost", + "basic_auth_enabled": True, + "basic_auth_user_attribute": generate_id(), + "basic_auth_password_attribute": generate_id(), + }, + ) + self.assertEqual(response.status_code, 201) + + def test_basic_auth_invalid(self): + """Test basic_auth_enabled""" + response = self.client.post( + reverse("authentik_api:proxyprovider-list"), + { + "name": generate_id(), + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + "internal_host": "http://localhost", + "basic_auth_enabled": True, + }, + ) + self.assertEqual(response.status_code, 400) + self.assertJSONEqual( + response.content.decode(), + { + "basic_auth_enabled": [ + "User and password attributes must be set when basic auth is enabled." + ] + }, + ) + + def test_validate(self): + """Test validate""" + response = self.client.post( + reverse("authentik_api:proxyprovider-list"), + { + "name": generate_id(), + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + }, + ) + self.assertEqual(response.status_code, 400) + self.assertJSONEqual( + response.content.decode(), + {"non_field_errors": ["Internal host cannot be empty when forward auth is disabled."]}, + ) + + def test_create_defaults(self): + """Test create""" + name = generate_id() + response = self.client.post( + reverse("authentik_api:proxyprovider-list"), + { + "name": name, + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + "internal_host": "http://localhost", + }, + ) + self.assertEqual(response.status_code, 201) + provider: ProxyProvider = ProxyProvider.objects.get(name=name) + self.assertEqual(provider.client_type, ClientTypes.CONFIDENTIAL) + + def test_update_defaults(self): + """Test create""" + name = generate_id() + response = self.client.post( + reverse("authentik_api:proxyprovider-list"), + { + "name": name, + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + "internal_host": "http://localhost", + }, + ) + self.assertEqual(response.status_code, 201) + provider: ProxyProvider = ProxyProvider.objects.get(name=name) + self.assertEqual(provider.client_type, ClientTypes.CONFIDENTIAL) + provider.client_type = ClientTypes.PUBLIC + provider.save() + response = self.client.put( + reverse("authentik_api:proxyprovider-detail", kwargs={"pk": provider.pk}), + { + "name": name, + "mode": ProxyMode.PROXY, + "authorization_flow": create_test_flow().pk.hex, + "external_host": "http://localhost", + "internal_host": "http://localhost", + }, + ) + self.assertEqual(response.status_code, 200) + provider: ProxyProvider = ProxyProvider.objects.get(name=name) + self.assertEqual(provider.client_type, ClientTypes.CONFIDENTIAL) diff --git a/tests/e2e/test_provider_ldap.py b/tests/e2e/test_provider_ldap.py index dae09b742..cc4dce79a 100644 --- a/tests/e2e/test_provider_ldap.py +++ b/tests/e2e/test_provider_ldap.py @@ -47,7 +47,6 @@ class TestProviderLDAP(SeleniumTestCase): def _prepare(self) -> User: """prepare user, provider, app and container""" - # set additionalHeaders to test later self.user.attributes["extraAttribute"] = "bar" self.user.save() diff --git a/tests/e2e/test_provider_proxy.py b/tests/e2e/test_provider_proxy.py index 5ce1683b7..7ea0bf359 100644 --- a/tests/e2e/test_provider_proxy.py +++ b/tests/e2e/test_provider_proxy.py @@ -1,4 +1,5 @@ """Proxy and Outpost e2e tests""" +from base64 import b64encode from dataclasses import asdict from sys import platform from time import sleep @@ -14,6 +15,7 @@ from authentik import __version__ from authentik.blueprints.tests import apply_blueprint, reconcile_app from authentik.core.models import Application from authentik.flows.models import Flow +from authentik.lib.generators import generate_id from authentik.outposts.models import DockerServiceConnection, Outpost, OutpostConfig, OutpostType from authentik.outposts.tasks import outpost_local_connection from authentik.providers.proxy.models import ProxyProvider @@ -119,6 +121,78 @@ class TestProviderProxy(SeleniumTestCase): full_body_text = self.driver.find_element(By.CSS_SELECTOR, ".pf-c-title.pf-m-3xl").text self.assertIn("You've logged out of proxy.", full_body_text) + @retry() + @apply_blueprint( + "default/10-flow-default-authentication-flow.yaml", + "default/10-flow-default-invalidation-flow.yaml", + ) + @apply_blueprint( + "default/20-flow-default-provider-authorization-explicit-consent.yaml", + "default/20-flow-default-provider-authorization-implicit-consent.yaml", + ) + @apply_blueprint( + "system/providers-oauth2.yaml", + "system/providers-proxy.yaml", + ) + @reconcile_app("authentik_crypto") + def test_proxy_basic_auth(self): + """Test simple outpost setup with single provider""" + cred = generate_id() + attr = "basic-password" # nosec + self.user.attributes["basic-username"] = cred + self.user.attributes[attr] = cred + self.user.save() + + proxy: ProxyProvider = ProxyProvider.objects.create( + name="proxy_provider", + authorization_flow=Flow.objects.get( + slug="default-provider-authorization-implicit-consent" + ), + internal_host="http://localhost", + external_host="http://localhost:9000", + basic_auth_enabled=True, + basic_auth_user_attribute="basic-username", + basic_auth_password_attribute=attr, + ) + # Ensure OAuth2 Params are set + proxy.set_oauth_defaults() + proxy.save() + # we need to create an application to actually access the proxy + Application.objects.create(name="proxy", slug="proxy", provider=proxy) + outpost: Outpost = Outpost.objects.create( + name="proxy_outpost", + type=OutpostType.PROXY, + ) + outpost.providers.add(proxy) + outpost.build_user_permissions(outpost.user) + + self.proxy_container = self.start_proxy(outpost) + + # Wait until outpost healthcheck succeeds + healthcheck_retries = 0 + while healthcheck_retries < 50: + if len(outpost.state) > 0: + state = outpost.state[0] + if state.last_seen: + break + healthcheck_retries += 1 + sleep(0.5) + sleep(5) + + self.driver.get("http://localhost:9000") + self.login() + sleep(1) + + full_body_text = self.driver.find_element(By.CSS_SELECTOR, "pre").text + self.assertIn(f"X-Authentik-Username: {self.user.username}", full_body_text) + auth_header = b64encode(f"{cred}:{cred}".encode()).decode() + self.assertIn(f"Authorization: Basic {auth_header}", full_body_text) + + self.driver.get("http://localhost:9000/outpost.goauthentik.io/sign_out") + sleep(2) + full_body_text = self.driver.find_element(By.CSS_SELECTOR, ".pf-c-title.pf-m-3xl").text + self.assertIn("You've logged out of proxy.", full_body_text) + @skipUnless(platform.startswith("linux"), "requires local docker") class TestProviderProxyConnect(ChannelsLiveServerTestCase):