stages: use managed secrets for Duo and SMS

authentik/rbac/tests/test_write_only_fields.py

@@ -1,19 +1,18 @@
 """Test write-only field exposure in API responses"""
 
-from json import loads
-
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
 from authentik.core.models import Group
 from authentik.core.tests.utils import create_test_admin_user, create_test_user
+from authentik.crypto.secrets.tests.utils import create_test_secret
 from authentik.lib.generators import generate_id
 from authentik.rbac.models import Role
 from authentik.stages.authenticator_sms.models import AuthenticatorSMSStage, SMSProviders
 
 
 class TestWriteOnlyFields(APITestCase):
-    """Test that write-only fields are never rendered, whatever the caller holds"""
+    """Credentials are never rendered by their consumers, whatever the caller holds"""
 
     def setUp(self) -> None:
         self.user = create_test_user()
@@ -23,113 +22,57 @@ def setUp(self) -> None:
         self.group.users.add(self.user)
 
         AuthenticatorSMSStage.objects.all().delete()
-        self.auth = generate_id()
-        self.auth_password = generate_id()
+        self.auth = create_test_secret(generate_id())
+        self.auth_password = create_test_secret(generate_id())
         self.stage = AuthenticatorSMSStage.objects.create(
             name=generate_id(),
             provider=SMSProviders.GENERIC,
             from_number=generate_id(),
-            account_sid=generate_id(),
-            auth=self.auth,
-            auth_password=self.auth_password,
+            auth_ref=self.auth,
+            auth_password_ref=self.auth_password,
         )
 
-    def test_stage_detail_view(self):
-        """Test stage detail (role has global view permission)"""
-        self.role.assign_perms("authentik_stages_authenticator_sms.view_authenticatorsmsstage")
-        self.client.force_login(self.user)
-
-        res = self.client.get(
-            reverse("authentik_api:authenticatorsmsstage-detail", kwargs={"pk": self.stage.pk})
-        )
+    def assert_hidden(self, url: str):
+        res = self.client.get(url)
         self.assertEqual(res.status_code, 200)
-        body = loads(res.content)
-        self.assertNotIn("auth", body)
-        self.assertNotIn("auth_password", body)
-        self.assertEqual(body["account_sid"], self.stage.account_sid)
+        for secret in [self.auth, self.auth_password]:
+            self.assertIn(str(secret.pk), res.content.decode())
+            self.assertNotIn(secret.secret_value, res.content.decode())
 
-    def test_stage_list_view(self):
-        """Test stage list (role has global view permission)"""
+    def test_detail_and_list(self):
+        """Values are hidden for users who can only view the stage"""
         self.role.assign_perms("authentik_stages_authenticator_sms.view_authenticatorsmsstage")
         self.client.force_login(self.user)
+        self.assert_hidden(
+            reverse(
+                "authentik_api:authenticatorsmsstage-detail",
+                kwargs={"pk": self.stage.pk},
+            )
+        )
+        self.assert_hidden(reverse("authentik_api:authenticatorsmsstage-list"))
 
-        res = self.client.get(reverse("authentik_api:authenticatorsmsstage-list"))
-        self.assertEqual(res.status_code, 200)
-        body = loads(res.content)
-        self.assertEqual(body["pagination"]["count"], 1)
-        self.assertNotIn("auth", body["results"][0])
-        self.assertNotIn("auth_password", body["results"][0])
-        self.assertEqual(body["results"][0]["account_sid"], self.stage.account_sid)
-
-    def test_stage_detail_change_global(self):
-        """Test stage detail (role has global change permission)"""
+    def test_change_permission(self):
+        """Values are hidden for users who can change the stage"""
         self.role.assign_perms(
             [
                 "authentik_stages_authenticator_sms.view_authenticatorsmsstage",
                 "authentik_stages_authenticator_sms.change_authenticatorsmsstage",
             ]
         )
         self.client.force_login(self.user)
-
-        res = self.client.get(
-            reverse("authentik_api:authenticatorsmsstage-detail", kwargs={"pk": self.stage.pk})
+        self.assert_hidden(
+            reverse(
+                "authentik_api:authenticatorsmsstage-detail",
+                kwargs={"pk": self.stage.pk},
+            )
         )
-        self.assertEqual(res.status_code, 200)
-        body = loads(res.content)
-        self.assertNotIn("auth", body)
-        self.assertNotIn("auth_password", body)
 
-    def test_stage_detail_superuser(self):
-        """Test stage detail (superuser)"""
+    def test_superuser(self):
+        """Values are hidden for superusers"""
         self.client.force_login(create_test_admin_user())
-
-        res = self.client.get(
-            reverse("authentik_api:authenticatorsmsstage-detail", kwargs={"pk": self.stage.pk})
+        self.assert_hidden(
+            reverse(
+                "authentik_api:authenticatorsmsstage-detail",
+                kwargs={"pk": self.stage.pk},
+            )
         )
-        self.assertEqual(res.status_code, 200)
-        body = loads(res.content)
-        self.assertNotIn("auth", body)
-        self.assertNotIn("auth_password", body)
-
-    def test_stage_create(self):
-        """Test stage create (the response omits the values that were stored)"""
-        self.role.assign_perms("authentik_stages_authenticator_sms.add_authenticatorsmsstage")
-        self.client.force_login(self.user)
-
-        name = generate_id()
-        auth = generate_id()
-        auth_password = generate_id()
-        res = self.client.post(
-            reverse("authentik_api:authenticatorsmsstage-list"),
-            {
-                "name": name,
-                "provider": SMSProviders.GENERIC,
-                "from_number": generate_id(),
-                "account_sid": generate_id(),
-                "auth": auth,
-                "auth_password": auth_password,
-            },
-        )
-        self.assertEqual(res.status_code, 201)
-        body = loads(res.content)
-        self.assertNotIn("auth", body)
-        self.assertNotIn("auth_password", body)
-        stage = AuthenticatorSMSStage.objects.get(name=name)
-        self.assertEqual(stage.auth, auth)
-        self.assertEqual(stage.auth_password, auth_password)
-
-    def test_stage_partial_update_keeps_values(self):
-        """Test stage partial update without the write-only fields (as the admin UI sends it)"""
-        self.client.force_login(create_test_admin_user())
-
-        from_number = generate_id()
-        res = self.client.patch(
-            reverse("authentik_api:authenticatorsmsstage-detail", kwargs={"pk": self.stage.pk}),
-            {"from_number": from_number},
-            content_type="application/json",
-        )
-        self.assertEqual(res.status_code, 200)
-        self.stage.refresh_from_db()
-        self.assertEqual(self.stage.from_number, from_number)
-        self.assertEqual(self.stage.auth, self.auth)
-        self.assertEqual(self.stage.auth_password, self.auth_password)

authentik/stages/authenticator_duo/api.py

@@ -41,15 +41,11 @@ class Meta:
             "configure_flow",
             "friendly_name",
             "client_id",
-            "client_secret",
+            "client_secret_ref",
             "api_hostname",
             "admin_integration_key",
-            "admin_secret_key",
+            "admin_secret_key_ref",
         ]
-        extra_kwargs = {
-            "client_secret": {"write_only": True},
-            "admin_secret_key": {"write_only": True},
-        }
 
 
 class AuthenticatorDuoStageManualDeviceImport(Serializer):
@@ -165,7 +161,7 @@ def import_device_manual(
     def import_devices_automatic(self, request: Request, pk: str) -> Response:
         """Import duo devices into authentik"""
         stage: AuthenticatorDuoStage = self.get_object()
-        if stage.admin_integration_key == "":
+        if not stage.admin_integration_key or not stage.admin_secret_key_ref_id:
             return Response(
                 data={
                     "non_field_errors": [
@@ -183,9 +179,6 @@ def _duo_import_devices(self, stage: AuthenticatorDuoStage) -> dict[str, Any]:
         Import duo devices. This used to be a blocking task.
         """
         created = 0
-        if stage.admin_integration_key == "":
-            LOGGER.info("Stage does not have admin integration configured", stage=stage)
-            return {"error": "Stage does not have admin integration configured", "count": created}
         client = stage.admin_client()
         try:
             for duo_user in client.get_users_iterator():

authentik/stages/authenticator_duo/migrations/0009_managed_secrets.py

@@ -0,0 +1,50 @@
+# Generated by Django 5.2.17 on 2026-10-10 19:25
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+from authentik.crypto.secrets.migrations._credential_values import move_credentials
+
+FIELDS = [
+    ("client_secret", "client_secret_ref", "text", "Duo client secret"),
+    ("admin_secret_key", "admin_secret_key_ref", "text", "Duo admin secret"),
+]
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("guardian", "0004_role_permissions"),
+        ("authentik_rbac", "0011_initialpermissionspermission"),
+        ("authentik_crypto_secrets", "0001_initial"),
+        ("authentik_stages_authenticator_duo", "0008_alter_duodevice_stage"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="authenticatorduostage",
+            name="admin_secret_key_ref",
+            field=models.ForeignKey(
+                blank=True,
+                default=None,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="duo_admin_secret_key_stages",
+                to="authentik_crypto_secrets.secret",
+                verbose_name="Admin secret key",
+            ),
+        ),
+        migrations.AddField(
+            model_name="authenticatorduostage",
+            name="client_secret_ref",
+            field=models.ForeignKey(
+                default=None,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                related_name="duo_client_secret_stages",
+                to="authentik_crypto_secrets.secret",
+                verbose_name="Client secret",
+            ),
+        ),
+        move_credentials("authentik_stages_authenticator_duo", "AuthenticatorDuoStage", FIELDS),
+    ]

authentik/stages/authenticator_duo/models.py

@@ -21,10 +21,29 @@ class AuthenticatorDuoStage(ConfigurableStage, FriendlyNamedStage, Stage):
     api_hostname = models.TextField()
 
     client_id = models.TextField()
+    # Legacy column, kept for downgrades. Remove in 2027.2.
     client_secret = models.TextField()
+    client_secret_ref = models.ForeignKey(
+        "authentik_crypto_secrets.Secret",
+        verbose_name=_("Client secret"),
+        on_delete=models.PROTECT,
+        null=True,
+        default=None,
+        related_name="duo_client_secret_stages",
+    )
 
     admin_integration_key = models.TextField(blank=True, default="")
+    # Legacy column, kept for downgrades. Remove in 2027.2.
     admin_secret_key = models.TextField(blank=True, default="")
+    admin_secret_key_ref = models.ForeignKey(
+        "authentik_crypto_secrets.Secret",
+        verbose_name=_("Admin secret key"),
+        on_delete=models.PROTECT,
+        null=True,
+        blank=True,
+        default=None,
+        related_name="duo_admin_secret_key_stages",
+    )
 
     @property
     def serializer(self) -> type[BaseSerializer]:
@@ -42,18 +61,18 @@ def auth_client(self) -> Auth:
         """Get an API Client to talk to duo"""
         return Auth(
             self.client_id,
-            self.client_secret,
+            self.client_secret_ref.secret_value,
             self.api_hostname,
             user_agent=authentik_user_agent(),
         )
 
     def admin_client(self) -> Admin:
         """Get an API Client to talk to duo"""
-        if self.admin_integration_key == "" or self.admin_secret_key == "":  # nosec
+        if self.admin_integration_key == "" or not self.admin_secret_key_ref:
             raise ValueError("Admin credentials not configured")
         client = Admin(
             self.admin_integration_key,
-            self.admin_secret_key,
+            self.admin_secret_key_ref.secret_value,
             self.api_hostname,
             user_agent=authentik_user_agent(),
         )