security: fix CVE 2024 52289 (#12113)

* initial migration

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* migrate tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix loading

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* start dynamic ui

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* initial ui

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* add serialize

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* add error message handling

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix/add tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* prepare docs

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* migrate to new input

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

---------

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

authentik/core/tests/test_applications_api.py

@@ -12,7 +12,7 @@
 from authentik.lib.generators import generate_id
 from authentik.policies.dummy.models import DummyPolicy
 from authentik.policies.models import PolicyBinding
-from authentik.providers.oauth2.models import OAuth2Provider
+from authentik.providers.oauth2.models import OAuth2Provider, RedirectURI, RedirectURIMatchingMode
 from authentik.providers.proxy.models import ProxyProvider
 from authentik.providers.saml.models import SAMLProvider
 
@@ -24,7 +24,7 @@ def setUp(self) -> None:
         self.user = create_test_admin_user()
         self.provider = OAuth2Provider.objects.create(
             name="test",
-            redirect_uris="http://some-other-domain",
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://some-other-domain")],
             authorization_flow=create_test_flow(),
         )
         self.allowed: Application = Application.objects.create(

authentik/core/tests/test_transactional_applications_api.py

@@ -35,6 +35,7 @@ def test_create_transactional(self):
                     "name": uid,
                     "authorization_flow": str(create_test_flow().pk),
                     "invalidation_flow": str(create_test_flow().pk),
+                    "redirect_uris": [],
                 },
             },
         )
@@ -89,6 +90,7 @@ def test_create_transactional_bindings(self):
                     "name": uid,
                     "authorization_flow": str(authorization_flow.pk),
                     "invalidation_flow": str(authorization_flow.pk),
+                    "redirect_uris": [],
                 },
                 "policy_bindings": [{"group": group.pk, "order": 0}],
             },
@@ -120,6 +122,7 @@ def test_create_transactional_invalid(self):
                     "name": uid,
                     "authorization_flow": "",
                     "invalidation_flow": "",
+                    "redirect_uris": [],
                 },
             },
         )

authentik/crypto/tests.py

@@ -18,7 +18,7 @@
 from authentik.crypto.tasks import MANAGED_DISCOVERED, certificate_discovery
 from authentik.lib.config import CONFIG
 from authentik.lib.generators import generate_id, generate_key
-from authentik.providers.oauth2.models import OAuth2Provider
+from authentik.providers.oauth2.models import OAuth2Provider, RedirectURI, RedirectURIMatchingMode
 
 
 class TestCrypto(APITestCase):
@@ -274,7 +274,7 @@ def test_used_by(self):
             client_id="test",
             client_secret=generate_key(),
             authorization_flow=create_test_flow(),
-            redirect_uris="http://localhost",
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
             signing_key=keypair,
         )
         response = self.client.get(
@@ -306,7 +306,7 @@ def test_used_by_denied(self):
             client_id="test",
             client_secret=generate_key(),
             authorization_flow=create_test_flow(),
-            redirect_uris="http://localhost",
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
             signing_key=keypair,
         )
         response = self.client.get(

authentik/providers/oauth2/api/providers.py

@@ -1,15 +1,18 @@
 """OAuth2Provider API Views"""
 
 from copy import copy
+from re import compile
+from re import error as RegexError
 
 from django.urls import reverse
 from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema
 from guardian.shortcuts import get_objects_for_user
 from rest_framework.decorators import action
 from rest_framework.exceptions import ValidationError
-from rest_framework.fields import CharField
+from rest_framework.fields import CharField, ChoiceField
 from rest_framework.generics import get_object_or_404
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -20,13 +23,39 @@
 from authentik.core.api.utils import PassiveSerializer, PropertyMappingPreviewSerializer
 from authentik.core.models import Provider
 from authentik.providers.oauth2.id_token import IDToken
-from authentik.providers.oauth2.models import AccessToken, OAuth2Provider, ScopeMapping
+from authentik.providers.oauth2.models import (
+    AccessToken,
+    OAuth2Provider,
+    RedirectURIMatchingMode,
+    ScopeMapping,
+)
 from authentik.rbac.decorators import permission_required
 
 
+class RedirectURISerializer(PassiveSerializer):
+    """A single allowed redirect URI entry"""
+
+    matching_mode = ChoiceField(choices=RedirectURIMatchingMode.choices)
+    url = CharField()
+
+
 class OAuth2ProviderSerializer(ProviderSerializer):
     """OAuth2Provider Serializer"""
 
+    redirect_uris = RedirectURISerializer(many=True, source="_redirect_uris")
+
+    def validate_redirect_uris(self, data: list) -> list:
+        for entry in data:
+            if entry.get("matching_mode") == RedirectURIMatchingMode.REGEX:
+                url = entry.get("url")
+                try:
+                    compile(url)
+                except RegexError:
+                    raise ValidationError(
+                        _("Invalid Regex Pattern: {url}".format(url=url))
+                    ) from None
+        return data
+
     class Meta:
         model = OAuth2Provider
         fields = ProviderSerializer.Meta.fields + [
@@ -79,7 +108,6 @@ class OAuth2ProviderViewSet(UsedByMixin, ModelViewSet):
         "refresh_token_validity",
         "include_claims_in_id_token",
         "signing_key",
-        "redirect_uris",
         "sub_mode",
         "property_mappings",
         "issuer_mode",

authentik/providers/oauth2/errors.py

@@ -7,7 +7,7 @@
 from authentik.events.models import Event, EventAction
 from authentik.lib.sentry import SentryIgnoredException
 from authentik.lib.views import bad_request_message
-from authentik.providers.oauth2.models import GrantTypes
+from authentik.providers.oauth2.models import GrantTypes, RedirectURI
 
 
 class OAuth2Error(SentryIgnoredException):
@@ -46,9 +46,9 @@ class RedirectUriError(OAuth2Error):
     )
 
     provided_uri: str
-    allowed_uris: list[str]
+    allowed_uris: list[RedirectURI]
 
-    def __init__(self, provided_uri: str, allowed_uris: list[str]) -> None:
+    def __init__(self, provided_uri: str, allowed_uris: list[RedirectURI]) -> None:
         super().__init__()
         self.provided_uri = provided_uri
         self.allowed_uris = allowed_uris

authentik/providers/oauth2/migrations/0024_remove_oauth2provider_redirect_uris_and_more.py

@@ -0,0 +1,48 @@
+# Generated by Django 5.0.9 on 2024-11-04 12:56
+from django.apps.registry import Apps
+
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+
+from django.db import migrations, models
+
+
+def migrate_redirect_uris(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
+    from authentik.providers.oauth2.models import RedirectURI, RedirectURIMatchingMode
+
+    OAuth2Provider = apps.get_model("authentik_providers_oauth2", "oauth2provider")
+
+    db_alias = schema_editor.connection.alias
+    for provider in OAuth2Provider.objects.using(db_alias).all():
+        uris = []
+        for old in provider.old_redirect_uris.split("\n"):
+            mode = RedirectURIMatchingMode.STRICT
+            if old == "*" or old == ".*":
+                mode = RedirectURIMatchingMode.REGEX
+            uris.append(RedirectURI(mode, url=old))
+        provider.redirect_uris = uris
+        provider.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_providers_oauth2", "0023_alter_accesstoken_refreshtoken_use_hash_index"),
+    ]
+
+    operations = [
+        migrations.RenameField(
+            model_name="oauth2provider",
+            old_name="redirect_uris",
+            new_name="old_redirect_uris",
+        ),
+        migrations.AddField(
+            model_name="oauth2provider",
+            name="_redirect_uris",
+            field=models.JSONField(default=dict, verbose_name="Redirect URIs"),
+        ),
+        migrations.RunPython(migrate_redirect_uris, lambda *args: ...),
+        migrations.RemoveField(
+            model_name="oauth2provider",
+            name="old_redirect_uris",
+        ),
+    ]

authentik/providers/oauth2/models.py

@@ -3,7 +3,7 @@
 import base64
 import binascii
 import json
-from dataclasses import asdict
+from dataclasses import asdict, dataclass
 from functools import cached_property
 from hashlib import sha256
 from typing import Any
@@ -12,6 +12,7 @@
 from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
 from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
+from dacite import Config
 from dacite.core import from_dict
 from django.contrib.postgres.indexes import HashIndex
 from django.db import models
@@ -77,11 +78,25 @@ class IssuerMode(models.TextChoices):
     """Configure how the `iss` field is created."""
 
     GLOBAL = "global", _("Same identifier is used for all providers")
-    PER_PROVIDER = "per_provider", _(
-        "Each provider has a different issuer, based on the application slug."
+    PER_PROVIDER = (
+        "per_provider",
+        _("Each provider has a different issuer, based on the application slug."),
     )
 
 
+class RedirectURIMatchingMode(models.TextChoices):
+    STRICT = "strict", _("Strict URL comparison")
+    REGEX = "regex", _("Regular Expression URL matching")
+
+
+@dataclass
+class RedirectURI:
+    """A single redirect URI entry"""
+
+    matching_mode: RedirectURIMatchingMode
+    url: str
+
+
 class ResponseTypes(models.TextChoices):
     """Response Type required by the client."""
 
@@ -156,11 +171,9 @@ class OAuth2Provider(WebfingerProvider, Provider):
         verbose_name=_("Client Secret"),
         default=generate_client_secret,
     )
-    redirect_uris = models.TextField(
-        default="",
-        blank=True,
+    _redirect_uris = models.JSONField(
+        default=dict,
         verbose_name=_("Redirect URIs"),
-        help_text=_("Enter each URI on a new line."),
     )
 
     include_claims_in_id_token = models.BooleanField(
@@ -271,12 +284,33 @@ def get_issuer(self, request: HttpRequest) -> str | None:
         except Provider.application.RelatedObjectDoesNotExist:
             return None
 
+    @property
+    def redirect_uris(self) -> list[RedirectURI]:
+        uris = []
+        for entry in self._redirect_uris:
+            uris.append(
+                from_dict(
+                    RedirectURI,
+                    entry,
+                    config=Config(type_hooks={RedirectURIMatchingMode: RedirectURIMatchingMode}),
+                )
+            )
+        return uris
+
+    @redirect_uris.setter
+    def redirect_uris(self, value: list[RedirectURI]):
+        cleansed = []
+        for entry in value:
+            cleansed.append(asdict(entry))
+        self._redirect_uris = cleansed
+
     @property
     def launch_url(self) -> str | None:
         """Guess launch_url based on first redirect_uri"""
-        if self.redirect_uris == "":
+        redirects = self.redirect_uris
+        if len(redirects) < 1:
             return None
-        main_url = self.redirect_uris.split("\n", maxsplit=1)[0]
+        main_url = redirects[0].url
         try:
             launch_url = urlparse(main_url)._replace(path="")
             return urlunparse(launch_url)

authentik/providers/oauth2/tests/test_api.py

@@ -10,7 +10,13 @@
 from authentik.blueprints.tests import apply_blueprint
 from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_admin_user, create_test_flow
-from authentik.providers.oauth2.models import OAuth2Provider, ScopeMapping
+from authentik.lib.generators import generate_id
+from authentik.providers.oauth2.models import (
+    OAuth2Provider,
+    RedirectURI,
+    RedirectURIMatchingMode,
+    ScopeMapping,
+)
 
 
 class TestAPI(APITestCase):
@@ -21,7 +27,7 @@ def setUp(self) -> None:
         self.provider: OAuth2Provider = OAuth2Provider.objects.create(
             name="test",
             authorization_flow=create_test_flow(),
-            redirect_uris="http://testserver",
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
         )
         self.provider.property_mappings.set(ScopeMapping.objects.all())
         self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
@@ -50,9 +56,29 @@ def test_setup_urls(self):
     @skipUnless(version_info >= (3, 11, 4), "This behaviour is only Python 3.11.4 and up")
     def test_launch_url(self):
         """Test launch_url"""
-        self.provider.redirect_uris = (
-            "https://[\\d\\w]+.pr.test.goauthentik.io/source/oauth/callback/authentik/\n"
-        )
+        self.provider.redirect_uris = [
+            RedirectURI(
+                RedirectURIMatchingMode.REGEX,
+                "https://[\\d\\w]+.pr.test.goauthentik.io/source/oauth/callback/authentik/",
+            ),
+        ]
         self.provider.save()
         self.provider.refresh_from_db()
         self.assertIsNone(self.provider.launch_url)
+
+    def test_validate_redirect_uris(self):
+        """Test redirect_uris API"""
+        response = self.client.post(
+            reverse("authentik_api:oauth2provider-list"),
+            data={
+                "name": generate_id(),
+                "authorization_flow": create_test_flow().pk,
+                "invalidation_flow": create_test_flow().pk,
+                "redirect_uris": [
+                    {"matching_mode": "strict", "url": "http://goauthentik.io"},
+                    {"matching_mode": "regex", "url": "**"},
+                ],
+            },
+        )
+        self.assertJSONEqual(response.content, {"redirect_uris": ["Invalid Regex Pattern: **"]})
+        self.assertEqual(response.status_code, 400)