authentik.recovery.tests

  1"""recovery tests"""
  2
  3from datetime import timedelta
  4from io import StringIO
  5
  6from django.core.management import call_command
  7from django.test import TestCase
  8from django.urls import reverse
  9from django.utils.timezone import now
 10from django_tenants.utils import get_public_schema_name
 11
 12from authentik.core.models import Token, TokenIntents, User
 13
 14
 15class TestRecovery(TestCase):
 16    """recovery tests"""
 17
 18    def setUp(self):
 19        self.user: User = User.objects.create_user(username="recovery-test-user")
 20
 21    def test_create_key(self):
 22        """Test creation of a new key"""
 23        out = StringIO()
 24        self.assertEqual(len(Token.objects.filter(intent=TokenIntents.INTENT_RECOVERY)), 0)
 25        call_command(
 26            "create_recovery_key",
 27            "5",
 28            self.user.username,
 29            schema=get_public_schema_name(),
 30            stdout=out,
 31        )
 32        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
 33        self.assertIn(token.key, out.getvalue())
 34        self.assertIn("valid for 5\xa0minutes", out.getvalue())
 35        self.assertEqual(len(Token.objects.filter(intent=TokenIntents.INTENT_RECOVERY)), 1)
 36
 37    def test_create_key_invalid(self):
 38        """Test creation of a new key (invalid)"""
 39        out = StringIO()
 40        self.assertEqual(len(Token.objects.filter(intent=TokenIntents.INTENT_RECOVERY)), 0)
 41        call_command("create_recovery_key", "5", "foo", schema=get_public_schema_name(), stderr=out)
 42        self.assertIn("not found", out.getvalue())
 43
 44    def test_recovery_view(self):
 45        """Test recovery view"""
 46        out = StringIO()
 47        call_command(
 48            "create_recovery_key",
 49            "10",
 50            self.user.username,
 51            schema=get_public_schema_name(),
 52            stdout=out,
 53        )
 54        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
 55        self.client.get(reverse("authentik_recovery:use-token", kwargs={"key": token.key}))
 56        self.assertEqual(self.client.session["authenticatedsession"].user.pk, token.user.pk)
 57
 58    def test_recovery_view_invalid(self):
 59        """Test recovery view with invalid token"""
 60        response = self.client.get(reverse("authentik_recovery:use-token", kwargs={"key": "abc"}))
 61        self.assertEqual(response.status_code, 404)
 62
 63    def test_recovery_admin_group_invalid(self):
 64        """Test creation of admin group"""
 65        out = StringIO()
 66        call_command("create_admin_group", "1", schema=get_public_schema_name(), stderr=out)
 67        self.assertIn("not found", out.getvalue())
 68
 69    def test_recovery_admin_group(self):
 70        """Test creation of admin group"""
 71        out = StringIO()
 72        call_command(
 73            "create_admin_group", self.user.username, schema=get_public_schema_name(), stdout=out
 74        )
 75        self.assertIn("successfully added to", out.getvalue())
 76        self.assertTrue(self.user.is_superuser)
 77
 78    def test_create_key_default_duration(self):
 79        """Test creation of a new key with default duration (60 minutes)"""
 80        out = StringIO()
 81        before_creation = now()
 82        call_command(
 83            "create_recovery_key",
 84            self.user.username,
 85            schema=get_public_schema_name(),
 86            stdout=out,
 87        )
 88        after_creation = now()
 89
 90        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
 91        self.assertIn(token.key, out.getvalue())
 92        self.assertIn("valid for 1\xa0hour", out.getvalue())
 93
 94        # Verify the token expires in approximately 60 minutes (default)
 95        expected_expiry_min = before_creation + timedelta(minutes=60)
 96        expected_expiry_max = after_creation + timedelta(minutes=60)
 97        self.assertGreaterEqual(token.expires, expected_expiry_min)
 98        self.assertLessEqual(token.expires, expected_expiry_max)
 99
100    def test_create_key_custom_duration(self):
101        """Test creation of a new key with custom duration"""
102        out = StringIO()
103        custom_duration = 120  # 2 hours
104        before_creation = now()
105
106        call_command(
107            "create_recovery_key",
108            str(custom_duration),
109            self.user.username,
110            schema=get_public_schema_name(),
111            stdout=out,
112        )
113        after_creation = now()
114
115        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
116        self.assertIn(token.key, out.getvalue())
117        self.assertIn("valid for 2\xa0hours", out.getvalue())
118
119        # Verify the token expires in approximately the custom duration
120        expected_expiry_min = before_creation + timedelta(minutes=custom_duration)
121        expected_expiry_max = after_creation + timedelta(minutes=custom_duration)
122        self.assertGreaterEqual(token.expires, expected_expiry_min)
123        self.assertLessEqual(token.expires, expected_expiry_max)
124
125    def test_create_key_short_duration(self):
126        """Test creation of a new key with very short duration (1 minute)"""
127        out = StringIO()
128        short_duration = 1
129        before_creation = now()
130
131        call_command(
132            "create_recovery_key",
133            str(short_duration),
134            self.user.username,
135            schema=get_public_schema_name(),
136            stdout=out,
137        )
138        after_creation = now()
139
140        token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
141        self.assertIn(token.key, out.getvalue())
142        self.assertIn("valid for 1\xa0minute", out.getvalue())
143
144        # Verify the token expires in approximately 1 minute
145        expected_expiry_min = before_creation + timedelta(minutes=short_duration)
146        expected_expiry_max = after_creation + timedelta(minutes=short_duration)
147        self.assertGreaterEqual(token.expires, expected_expiry_min)
148        self.assertLessEqual(token.expires, expected_expiry_max)
149
150    def test_create_key_duration_validation(self):
151        """Test that the duration is correctly converted to minutes"""
152        # Test various durations to ensure they're calculated correctly
153        test_cases = [1, 5, 30, 60, 120, 1440]  # 1min, 5min, 30min, 1hr, 2hr, 24hr
154
155        for duration in test_cases:
156            with self.subTest(duration=duration):
157                out = StringIO()
158                before_creation = now()
159
160                call_command(
161                    "create_recovery_key",
162                    str(duration),
163                    self.user.username,
164                    schema=get_public_schema_name(),
165                    stdout=out,
166                )
167                after_creation = now()
168
169                token = Token.objects.get(intent=TokenIntents.INTENT_RECOVERY, user=self.user)
170
171                # Verify the token expires in approximately the specified duration
172                expected_expiry_min = before_creation + timedelta(minutes=duration)
173                expected_expiry_max = after_creation + timedelta(minutes=duration)
174                self.assertGreaterEqual(token.expires, expected_expiry_min)
175                self.assertLessEqual(token.expires, expected_expiry_max)
176
177                # Clean up for next iteration
178                token.delete()
179
180    def test_create_key_help_text(self):
181        """Test that the help text correctly indicates minutes"""
182        from authentik.recovery.management.commands.create_recovery_key import Command
183
184        command = Command()
185        # Check that the help text mentions minutes
186        parser = command.create_parser("test", "create_recovery_key")
187        help_text = parser.format_help()
188        self.assertIn("minutes", help_text.lower())
189        self.assertNotIn("years", help_text.lower())