authentik.admin.files.backends.tests.test_s3_backend 

  1from unittest import skipUnless
  2from urllib.parse import parse_qs, urlsplit
  3
  4from botocore.exceptions import UnsupportedSignatureVersionError
  5from django.test import TestCase
  6
  7from authentik.admin.files.tests.utils import FileTestS3BackendMixin, s3_test_server_available
  8from authentik.admin.files.usage import FileUsage
  9from authentik.lib.config import CONFIG
 10
 11
 12@skipUnless(s3_test_server_available(), "S3 test server not available")
 13class TestS3Backend(FileTestS3BackendMixin, TestCase):
 14    """Test S3 backend functionality"""
 15
 16    def setUp(self):
 17        super().setUp()
 18
 19    def test_base_path(self):
 20        """Test base_path property generates correct S3 key prefix"""
 21        expected = "media/public"
 22        self.assertEqual(self.media_s3_backend.base_path, expected)
 23
 24    def test_supports_file_path_s3(self):
 25        """Test supports_file_path returns True for s3 backend"""
 26        self.assertTrue(self.media_s3_backend.supports_file("path/to/any-file.png"))
 27        self.assertTrue(self.media_s3_backend.supports_file("any-file.png"))
 28
 29    def test_list_files(self):
 30        """Test list_files returns relative paths"""
 31        self.media_s3_backend.client.put_object(
 32            Bucket=self.media_s3_bucket_name,
 33            Key="media/public/file1.png",
 34            Body=b"test content",
 35            ACL="private",
 36        )
 37        self.media_s3_backend.client.put_object(
 38            Bucket=self.media_s3_bucket_name,
 39            Key="media/other/file1.png",
 40            Body=b"test content",
 41            ACL="private",
 42        )
 43
 44        files = list(self.media_s3_backend.list_files())
 45
 46        self.assertEqual(len(files), 1)
 47        self.assertIn("file1.png", files)
 48
 49    def test_list_files_empty(self):
 50        """Test list_files with no files"""
 51        files = list(self.media_s3_backend.list_files())
 52
 53        self.assertEqual(len(files), 0)
 54
 55    def test_save_file(self):
 56        """Test save_file uploads to S3"""
 57        content = b"test file content"
 58        self.media_s3_backend.save_file("test.png", content)
 59
 60    def test_save_file_stream(self):
 61        """Test save_file_stream uploads to S3 using context manager"""
 62        with self.media_s3_backend.save_file_stream("test.csv") as f:
 63            f.write(b"header1,header2\n")
 64            f.write(b"value1,value2\n")
 65
 66    def test_delete_file(self):
 67        """Test delete_file removes from S3"""
 68        self.media_s3_backend.client.put_object(
 69            Bucket=self.media_s3_bucket_name,
 70            Key="media/public/test.png",
 71            Body=b"test content",
 72            ACL="private",
 73        )
 74        self.media_s3_backend.delete_file("test.png")
 75
 76    @CONFIG.patch("storage.s3.secure_urls", True)
 77    @CONFIG.patch("storage.s3.custom_domain", None)
 78    def test_file_url_basic(self):
 79        """Test file_url generates presigned URL with AWS signature format"""
 80        url = self.media_s3_backend.file_url("test.png")
 81
 82        self.assertIn("X-Amz-Algorithm=AWS4-HMAC-SHA256", url)
 83        self.assertIn("X-Amz-Signature=", url)
 84        self.assertIn("test.png", url)
 85
 86    def test_client_signature_version_default_v4(self):
 87        """Test S3 client defaults to v4 signature when not configured."""
 88        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3v4")
 89
 90    @CONFIG.patch("storage.s3.signature_version", "s3")
 91    def test_client_signature_version_global_override(self):
 92        """Test S3 client respects globally configured signature version."""
 93        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")
 94
 95    @CONFIG.patch("storage.s3.signature_version", "s3v4")
 96    @CONFIG.patch("storage.media.s3.signature_version", "s3")
 97    def test_client_signature_version_media_override(self):
 98        """Test usage-specific signature version takes precedence over global."""
 99        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")
100
101    @CONFIG.patch("storage.media.s3.signature_version", "not-a-real-signature")
102    def test_client_signature_version_unsupported(self):
103        """Test unsupported signature version raises botocore error."""
104        with self.assertRaises(UnsupportedSignatureVersionError):
105            self.media_s3_backend.file_url("test.png", use_cache=False)
106
107    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
108    def test_file_exists_true(self):
109        """Test file_exists returns True for existing file"""
110        self.media_s3_backend.client.put_object(
111            Bucket=self.media_s3_bucket_name,
112            Key="media/public/test.png",
113            Body=b"test content",
114            ACL="private",
115        )
116
117        exists = self.media_s3_backend.file_exists("test.png")
118
119        self.assertTrue(exists)
120
121    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
122    def test_file_exists_false(self):
123        """Test file_exists returns False for non-existent file"""
124        exists = self.media_s3_backend.file_exists("nonexistent.png")
125
126        self.assertFalse(exists)
127
128    def test_allowed_usages(self):
129        """Test that S3Backend supports all usage types"""
130        self.assertEqual(self.media_s3_backend.allowed_usages, list(FileUsage))
131
132    def test_reports_usage(self):
133        """Test S3Backend with REPORTS usage"""
134        self.assertEqual(self.reports_s3_backend.usage, FileUsage.REPORTS)
135        self.assertEqual(self.reports_s3_backend.base_path, "reports/public")
136
137    @CONFIG.patch("storage.s3.secure_urls", True)
138    @CONFIG.patch("storage.s3.addressing_style", "path")
139    def test_file_url_custom_domain_with_bucket_no_duplicate(self):
140        """Test file_url doesn't duplicate bucket name when custom_domain includes bucket.
141
142        Regression test for https://github.com/goauthentik/authentik/issues/19521
143
144        When using:
145        - Path-style addressing (bucket name goes in URL path, not subdomain)
146        - Custom domain that includes the bucket name (e.g., s3.example.com/bucket-name)
147
148        The bucket name should NOT appear twice in the final URL.
149
150        Example of the bug:
151        - custom_domain = "s3.example.com/authentik-media"
152        - boto3 presigned URL = "http://s3.example.com/authentik-media/media/public/file.png?..."
153        - Buggy result = "https://s3.example.com/authentik-media/authentik-media/media/public/file.png?..."
154        """
155        bucket_name = self.media_s3_bucket_name
156
157        # Custom domain includes the bucket name
158        custom_domain = f"localhost:8020/{bucket_name}"
159
160        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
161            url = self.media_s3_backend.file_url("application-icons/test.svg", use_cache=False)
162
163        # The bucket name should appear exactly once in the URL path, not twice
164        bucket_occurrences = url.count(bucket_name)
165        self.assertEqual(
166            bucket_occurrences,
167            1,
168            f"Bucket name '{bucket_name}' appears {bucket_occurrences} times in URL, expected 1. "
169            f"URL: {url}",
170        )
171
172    @CONFIG.patch("storage.s3.secure_urls", False)
173    @CONFIG.patch("storage.s3.addressing_style", "path")
174    def test_file_url_custom_domain_resigns_for_custom_host(self):
175        """Test presigned URLs are signed for the custom domain host.
176
177        Host-changing custom domains must produce a signature query string for
178        the public host, not reuse the internal endpoint signature.
179        """
180        bucket_name = self.media_s3_bucket_name
181        key_name = "application-icons/test.svg"
182        custom_domain = f"files.example.test:8020/{bucket_name}"
183
184        endpoint_signed_url = self.media_s3_backend.client.generate_presigned_url(
185            "get_object",
186            Params={
187                "Bucket": bucket_name,
188                "Key": f"{self.media_s3_backend.base_path}/{key_name}",
189            },
190            ExpiresIn=900,
191            HttpMethod="GET",
192        )
193
194        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
195            custom_url = self.media_s3_backend.file_url(key_name, use_cache=False)
196
197        endpoint_parts = urlsplit(endpoint_signed_url)
198        custom_parts = urlsplit(custom_url)
199
200        self.assertEqual(custom_parts.scheme, "http")
201        self.assertEqual(custom_parts.netloc, "files.example.test:8020")
202        self.assertEqual(parse_qs(custom_parts.query)["X-Amz-SignedHeaders"], ["host"])
203        self.assertNotEqual(
204            custom_parts.query,
205            endpoint_parts.query,
206            "Custom-domain URLs must be signed for the public host, not reuse the endpoint "
207            "signature query string.",
208        )
209
210    def test_themed_urls_without_theme_variable(self):
211        """Test themed_urls returns None when filename has no %(theme)s"""
212        result = self.media_s3_backend.themed_urls("logo.png")
213        self.assertIsNone(result)
214
215    def test_themed_urls_with_theme_variable(self):
216        """Test themed_urls returns dict of presigned URLs for each theme"""
217        result = self.media_s3_backend.themed_urls("logo-%(theme)s.png")
218
219        self.assertIsInstance(result, dict)
220        self.assertIn("light", result)
221        self.assertIn("dark", result)
222
223        # Check URLs are valid presigned URLs with correct file paths
224        self.assertIn("logo-light.png", result["light"])
225        self.assertIn("logo-dark.png", result["dark"])
226        self.assertIn("X-Amz-Signature=", result["light"])
227        self.assertIn("X-Amz-Signature=", result["dark"])
228
229    def test_themed_urls_multiple_theme_variables(self):
230        """Test themed_urls with multiple %(theme)s in path"""
231        result = self.media_s3_backend.themed_urls("%(theme)s/logo-%(theme)s.svg")
232
233        self.assertIsInstance(result, dict)
234        self.assertIn("light/logo-light.svg", result["light"])
235        self.assertIn("dark/logo-dark.svg", result["dark"])
236
237    def test_save_file_sets_content_type_svg(self):
238        """Test save_file sets correct ContentType for SVG files"""
239        self.media_s3_backend.save_file("test.svg", b"<svg></svg>")
240
241        response = self.media_s3_backend.client.head_object(
242            Bucket=self.media_s3_bucket_name,
243            Key="media/public/test.svg",
244        )
245        self.assertEqual(response["ContentType"], "image/svg+xml")
246
247    def test_save_file_sets_content_type_png(self):
248        """Test save_file sets correct ContentType for PNG files"""
249        self.media_s3_backend.save_file("test.png", b"\x89PNG\r\n\x1a\n")
250
251        response = self.media_s3_backend.client.head_object(
252            Bucket=self.media_s3_bucket_name,
253            Key="media/public/test.png",
254        )
255        self.assertEqual(response["ContentType"], "image/png")
256
257    def test_save_file_stream_sets_content_type(self):
258        """Test save_file_stream sets correct ContentType"""
259        with self.media_s3_backend.save_file_stream("test.css") as f:
260            f.write(b"body { color: red; }")
261
262        response = self.media_s3_backend.client.head_object(
263            Bucket=self.media_s3_bucket_name,
264            Key="media/public/test.css",
265        )
266        self.assertEqual(response["ContentType"], "text/css")
267
268    def test_save_file_unknown_extension_octet_stream(self):
269        """Test save_file sets octet-stream for unknown extensions"""
270        self.media_s3_backend.save_file("test.unknownext123", b"data")
271
272        response = self.media_s3_backend.client.head_object(
273            Bucket=self.media_s3_bucket_name,
274            Key="media/public/test.unknownext123",
275        )
276        self.assertEqual(response["ContentType"], "application/octet-stream")
@skipUnless(s3_test_server_available(), 'S3 test server not available')
class TestS3Backend(authentik.admin.files.tests.utils.FileTestS3BackendMixin, django.test.testcases.TestCase): 
 13@skipUnless(s3_test_server_available(), "S3 test server not available")
 14class TestS3Backend(FileTestS3BackendMixin, TestCase):
 15    """Test S3 backend functionality"""
 16
 17    def setUp(self):
 18        super().setUp()
 19
 20    def test_base_path(self):
 21        """Test base_path property generates correct S3 key prefix"""
 22        expected = "media/public"
 23        self.assertEqual(self.media_s3_backend.base_path, expected)
 24
 25    def test_supports_file_path_s3(self):
 26        """Test supports_file_path returns True for s3 backend"""
 27        self.assertTrue(self.media_s3_backend.supports_file("path/to/any-file.png"))
 28        self.assertTrue(self.media_s3_backend.supports_file("any-file.png"))
 29
 30    def test_list_files(self):
 31        """Test list_files returns relative paths"""
 32        self.media_s3_backend.client.put_object(
 33            Bucket=self.media_s3_bucket_name,
 34            Key="media/public/file1.png",
 35            Body=b"test content",
 36            ACL="private",
 37        )
 38        self.media_s3_backend.client.put_object(
 39            Bucket=self.media_s3_bucket_name,
 40            Key="media/other/file1.png",
 41            Body=b"test content",
 42            ACL="private",
 43        )
 44
 45        files = list(self.media_s3_backend.list_files())
 46
 47        self.assertEqual(len(files), 1)
 48        self.assertIn("file1.png", files)
 49
 50    def test_list_files_empty(self):
 51        """Test list_files with no files"""
 52        files = list(self.media_s3_backend.list_files())
 53
 54        self.assertEqual(len(files), 0)
 55
 56    def test_save_file(self):
 57        """Test save_file uploads to S3"""
 58        content = b"test file content"
 59        self.media_s3_backend.save_file("test.png", content)
 60
 61    def test_save_file_stream(self):
 62        """Test save_file_stream uploads to S3 using context manager"""
 63        with self.media_s3_backend.save_file_stream("test.csv") as f:
 64            f.write(b"header1,header2\n")
 65            f.write(b"value1,value2\n")
 66
 67    def test_delete_file(self):
 68        """Test delete_file removes from S3"""
 69        self.media_s3_backend.client.put_object(
 70            Bucket=self.media_s3_bucket_name,
 71            Key="media/public/test.png",
 72            Body=b"test content",
 73            ACL="private",
 74        )
 75        self.media_s3_backend.delete_file("test.png")
 76
 77    @CONFIG.patch("storage.s3.secure_urls", True)
 78    @CONFIG.patch("storage.s3.custom_domain", None)
 79    def test_file_url_basic(self):
 80        """Test file_url generates presigned URL with AWS signature format"""
 81        url = self.media_s3_backend.file_url("test.png")
 82
 83        self.assertIn("X-Amz-Algorithm=AWS4-HMAC-SHA256", url)
 84        self.assertIn("X-Amz-Signature=", url)
 85        self.assertIn("test.png", url)
 86
 87    def test_client_signature_version_default_v4(self):
 88        """Test S3 client defaults to v4 signature when not configured."""
 89        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3v4")
 90
 91    @CONFIG.patch("storage.s3.signature_version", "s3")
 92    def test_client_signature_version_global_override(self):
 93        """Test S3 client respects globally configured signature version."""
 94        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")
 95
 96    @CONFIG.patch("storage.s3.signature_version", "s3v4")
 97    @CONFIG.patch("storage.media.s3.signature_version", "s3")
 98    def test_client_signature_version_media_override(self):
 99        """Test usage-specific signature version takes precedence over global."""
100        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")
101
102    @CONFIG.patch("storage.media.s3.signature_version", "not-a-real-signature")
103    def test_client_signature_version_unsupported(self):
104        """Test unsupported signature version raises botocore error."""
105        with self.assertRaises(UnsupportedSignatureVersionError):
106            self.media_s3_backend.file_url("test.png", use_cache=False)
107
108    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
109    def test_file_exists_true(self):
110        """Test file_exists returns True for existing file"""
111        self.media_s3_backend.client.put_object(
112            Bucket=self.media_s3_bucket_name,
113            Key="media/public/test.png",
114            Body=b"test content",
115            ACL="private",
116        )
117
118        exists = self.media_s3_backend.file_exists("test.png")
119
120        self.assertTrue(exists)
121
122    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
123    def test_file_exists_false(self):
124        """Test file_exists returns False for non-existent file"""
125        exists = self.media_s3_backend.file_exists("nonexistent.png")
126
127        self.assertFalse(exists)
128
129    def test_allowed_usages(self):
130        """Test that S3Backend supports all usage types"""
131        self.assertEqual(self.media_s3_backend.allowed_usages, list(FileUsage))
132
133    def test_reports_usage(self):
134        """Test S3Backend with REPORTS usage"""
135        self.assertEqual(self.reports_s3_backend.usage, FileUsage.REPORTS)
136        self.assertEqual(self.reports_s3_backend.base_path, "reports/public")
137
138    @CONFIG.patch("storage.s3.secure_urls", True)
139    @CONFIG.patch("storage.s3.addressing_style", "path")
140    def test_file_url_custom_domain_with_bucket_no_duplicate(self):
141        """Test file_url doesn't duplicate bucket name when custom_domain includes bucket.
142
143        Regression test for https://github.com/goauthentik/authentik/issues/19521
144
145        When using:
146        - Path-style addressing (bucket name goes in URL path, not subdomain)
147        - Custom domain that includes the bucket name (e.g., s3.example.com/bucket-name)
148
149        The bucket name should NOT appear twice in the final URL.
150
151        Example of the bug:
152        - custom_domain = "s3.example.com/authentik-media"
153        - boto3 presigned URL = "http://s3.example.com/authentik-media/media/public/file.png?..."
154        - Buggy result = "https://s3.example.com/authentik-media/authentik-media/media/public/file.png?..."
155        """
156        bucket_name = self.media_s3_bucket_name
157
158        # Custom domain includes the bucket name
159        custom_domain = f"localhost:8020/{bucket_name}"
160
161        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
162            url = self.media_s3_backend.file_url("application-icons/test.svg", use_cache=False)
163
164        # The bucket name should appear exactly once in the URL path, not twice
165        bucket_occurrences = url.count(bucket_name)
166        self.assertEqual(
167            bucket_occurrences,
168            1,
169            f"Bucket name '{bucket_name}' appears {bucket_occurrences} times in URL, expected 1. "
170            f"URL: {url}",
171        )
172
173    @CONFIG.patch("storage.s3.secure_urls", False)
174    @CONFIG.patch("storage.s3.addressing_style", "path")
175    def test_file_url_custom_domain_resigns_for_custom_host(self):
176        """Test presigned URLs are signed for the custom domain host.
177
178        Host-changing custom domains must produce a signature query string for
179        the public host, not reuse the internal endpoint signature.
180        """
181        bucket_name = self.media_s3_bucket_name
182        key_name = "application-icons/test.svg"
183        custom_domain = f"files.example.test:8020/{bucket_name}"
184
185        endpoint_signed_url = self.media_s3_backend.client.generate_presigned_url(
186            "get_object",
187            Params={
188                "Bucket": bucket_name,
189                "Key": f"{self.media_s3_backend.base_path}/{key_name}",
190            },
191            ExpiresIn=900,
192            HttpMethod="GET",
193        )
194
195        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
196            custom_url = self.media_s3_backend.file_url(key_name, use_cache=False)
197
198        endpoint_parts = urlsplit(endpoint_signed_url)
199        custom_parts = urlsplit(custom_url)
200
201        self.assertEqual(custom_parts.scheme, "http")
202        self.assertEqual(custom_parts.netloc, "files.example.test:8020")
203        self.assertEqual(parse_qs(custom_parts.query)["X-Amz-SignedHeaders"], ["host"])
204        self.assertNotEqual(
205            custom_parts.query,
206            endpoint_parts.query,
207            "Custom-domain URLs must be signed for the public host, not reuse the endpoint "
208            "signature query string.",
209        )
210
211    def test_themed_urls_without_theme_variable(self):
212        """Test themed_urls returns None when filename has no %(theme)s"""
213        result = self.media_s3_backend.themed_urls("logo.png")
214        self.assertIsNone(result)
215
216    def test_themed_urls_with_theme_variable(self):
217        """Test themed_urls returns dict of presigned URLs for each theme"""
218        result = self.media_s3_backend.themed_urls("logo-%(theme)s.png")
219
220        self.assertIsInstance(result, dict)
221        self.assertIn("light", result)
222        self.assertIn("dark", result)
223
224        # Check URLs are valid presigned URLs with correct file paths
225        self.assertIn("logo-light.png", result["light"])
226        self.assertIn("logo-dark.png", result["dark"])
227        self.assertIn("X-Amz-Signature=", result["light"])
228        self.assertIn("X-Amz-Signature=", result["dark"])
229
230    def test_themed_urls_multiple_theme_variables(self):
231        """Test themed_urls with multiple %(theme)s in path"""
232        result = self.media_s3_backend.themed_urls("%(theme)s/logo-%(theme)s.svg")
233
234        self.assertIsInstance(result, dict)
235        self.assertIn("light/logo-light.svg", result["light"])
236        self.assertIn("dark/logo-dark.svg", result["dark"])
237
238    def test_save_file_sets_content_type_svg(self):
239        """Test save_file sets correct ContentType for SVG files"""
240        self.media_s3_backend.save_file("test.svg", b"<svg></svg>")
241
242        response = self.media_s3_backend.client.head_object(
243            Bucket=self.media_s3_bucket_name,
244            Key="media/public/test.svg",
245        )
246        self.assertEqual(response["ContentType"], "image/svg+xml")
247
248    def test_save_file_sets_content_type_png(self):
249        """Test save_file sets correct ContentType for PNG files"""
250        self.media_s3_backend.save_file("test.png", b"\x89PNG\r\n\x1a\n")
251
252        response = self.media_s3_backend.client.head_object(
253            Bucket=self.media_s3_bucket_name,
254            Key="media/public/test.png",
255        )
256        self.assertEqual(response["ContentType"], "image/png")
257
258    def test_save_file_stream_sets_content_type(self):
259        """Test save_file_stream sets correct ContentType"""
260        with self.media_s3_backend.save_file_stream("test.css") as f:
261            f.write(b"body { color: red; }")
262
263        response = self.media_s3_backend.client.head_object(
264            Bucket=self.media_s3_bucket_name,
265            Key="media/public/test.css",
266        )
267        self.assertEqual(response["ContentType"], "text/css")
268
269    def test_save_file_unknown_extension_octet_stream(self):
270        """Test save_file sets octet-stream for unknown extensions"""
271        self.media_s3_backend.save_file("test.unknownext123", b"data")
272
273        response = self.media_s3_backend.client.head_object(
274            Bucket=self.media_s3_bucket_name,
275            Key="media/public/test.unknownext123",
276        )
277        self.assertEqual(response["ContentType"], "application/octet-stream")

Test S3 backend functionality

def setUp(self): 
17    def setUp(self):
18        super().setUp()

Hook method for setting up the test fixture before exercising it.

def test_base_path(self): 
20    def test_base_path(self):
21        """Test base_path property generates correct S3 key prefix"""
22        expected = "media/public"
23        self.assertEqual(self.media_s3_backend.base_path, expected)

Test base_path property generates correct S3 key prefix

def test_supports_file_path_s3(self): 
25    def test_supports_file_path_s3(self):
26        """Test supports_file_path returns True for s3 backend"""
27        self.assertTrue(self.media_s3_backend.supports_file("path/to/any-file.png"))
28        self.assertTrue(self.media_s3_backend.supports_file("any-file.png"))

Test supports_file_path returns True for s3 backend

def test_list_files(self): 
30    def test_list_files(self):
31        """Test list_files returns relative paths"""
32        self.media_s3_backend.client.put_object(
33            Bucket=self.media_s3_bucket_name,
34            Key="media/public/file1.png",
35            Body=b"test content",
36            ACL="private",
37        )
38        self.media_s3_backend.client.put_object(
39            Bucket=self.media_s3_bucket_name,
40            Key="media/other/file1.png",
41            Body=b"test content",
42            ACL="private",
43        )
44
45        files = list(self.media_s3_backend.list_files())
46
47        self.assertEqual(len(files), 1)
48        self.assertIn("file1.png", files)

Test list_files returns relative paths

def test_list_files_empty(self): 
50    def test_list_files_empty(self):
51        """Test list_files with no files"""
52        files = list(self.media_s3_backend.list_files())
53
54        self.assertEqual(len(files), 0)

Test list_files with no files

def test_save_file(self): 
56    def test_save_file(self):
57        """Test save_file uploads to S3"""
58        content = b"test file content"
59        self.media_s3_backend.save_file("test.png", content)

Test save_file uploads to S3

def test_save_file_stream(self): 
61    def test_save_file_stream(self):
62        """Test save_file_stream uploads to S3 using context manager"""
63        with self.media_s3_backend.save_file_stream("test.csv") as f:
64            f.write(b"header1,header2\n")
65            f.write(b"value1,value2\n")

Test save_file_stream uploads to S3 using context manager

def test_delete_file(self): 
67    def test_delete_file(self):
68        """Test delete_file removes from S3"""
69        self.media_s3_backend.client.put_object(
70            Bucket=self.media_s3_bucket_name,
71            Key="media/public/test.png",
72            Body=b"test content",
73            ACL="private",
74        )
75        self.media_s3_backend.delete_file("test.png")

Test delete_file removes from S3

@CONFIG.patch('storage.s3.secure_urls', True)
@CONFIG.patch('storage.s3.custom_domain', None)
def test_file_url_basic(self): 
77    @CONFIG.patch("storage.s3.secure_urls", True)
78    @CONFIG.patch("storage.s3.custom_domain", None)
79    def test_file_url_basic(self):
80        """Test file_url generates presigned URL with AWS signature format"""
81        url = self.media_s3_backend.file_url("test.png")
82
83        self.assertIn("X-Amz-Algorithm=AWS4-HMAC-SHA256", url)
84        self.assertIn("X-Amz-Signature=", url)
85        self.assertIn("test.png", url)

Test file_url generates presigned URL with AWS signature format

def test_client_signature_version_default_v4(self): 
87    def test_client_signature_version_default_v4(self):
88        """Test S3 client defaults to v4 signature when not configured."""
89        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3v4")

Test S3 client defaults to v4 signature when not configured.

@CONFIG.patch('storage.s3.signature_version', 's3')
def test_client_signature_version_global_override(self): 
91    @CONFIG.patch("storage.s3.signature_version", "s3")
92    def test_client_signature_version_global_override(self):
93        """Test S3 client respects globally configured signature version."""
94        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")

Test S3 client respects globally configured signature version.

@CONFIG.patch('storage.s3.signature_version', 's3v4')
@CONFIG.patch('storage.media.s3.signature_version', 's3')
def test_client_signature_version_media_override(self): 
 96    @CONFIG.patch("storage.s3.signature_version", "s3v4")
 97    @CONFIG.patch("storage.media.s3.signature_version", "s3")
 98    def test_client_signature_version_media_override(self):
 99        """Test usage-specific signature version takes precedence over global."""
100        self.assertEqual(self.media_s3_backend.client.meta.config.signature_version, "s3")

Test usage-specific signature version takes precedence over global.

@CONFIG.patch('storage.media.s3.signature_version', 'not-a-real-signature')
def test_client_signature_version_unsupported(self): 
102    @CONFIG.patch("storage.media.s3.signature_version", "not-a-real-signature")
103    def test_client_signature_version_unsupported(self):
104        """Test unsupported signature version raises botocore error."""
105        with self.assertRaises(UnsupportedSignatureVersionError):
106            self.media_s3_backend.file_url("test.png", use_cache=False)

Test unsupported signature version raises botocore error.

@CONFIG.patch('storage.s3.bucket_name', 'test-bucket')
def test_file_exists_true(self): 
108    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
109    def test_file_exists_true(self):
110        """Test file_exists returns True for existing file"""
111        self.media_s3_backend.client.put_object(
112            Bucket=self.media_s3_bucket_name,
113            Key="media/public/test.png",
114            Body=b"test content",
115            ACL="private",
116        )
117
118        exists = self.media_s3_backend.file_exists("test.png")
119
120        self.assertTrue(exists)

Test file_exists returns True for existing file

@CONFIG.patch('storage.s3.bucket_name', 'test-bucket')
def test_file_exists_false(self): 
122    @CONFIG.patch("storage.s3.bucket_name", "test-bucket")
123    def test_file_exists_false(self):
124        """Test file_exists returns False for non-existent file"""
125        exists = self.media_s3_backend.file_exists("nonexistent.png")
126
127        self.assertFalse(exists)

Test file_exists returns False for non-existent file

def test_allowed_usages(self): 
129    def test_allowed_usages(self):
130        """Test that S3Backend supports all usage types"""
131        self.assertEqual(self.media_s3_backend.allowed_usages, list(FileUsage))

Test that S3Backend supports all usage types

def test_reports_usage(self): 
133    def test_reports_usage(self):
134        """Test S3Backend with REPORTS usage"""
135        self.assertEqual(self.reports_s3_backend.usage, FileUsage.REPORTS)
136        self.assertEqual(self.reports_s3_backend.base_path, "reports/public")

Test S3Backend with REPORTS usage

@CONFIG.patch('storage.s3.secure_urls', True)
@CONFIG.patch('storage.s3.addressing_style', 'path')
def test_file_url_custom_domain_with_bucket_no_duplicate(self): 
138    @CONFIG.patch("storage.s3.secure_urls", True)
139    @CONFIG.patch("storage.s3.addressing_style", "path")
140    def test_file_url_custom_domain_with_bucket_no_duplicate(self):
141        """Test file_url doesn't duplicate bucket name when custom_domain includes bucket.
142
143        Regression test for https://github.com/goauthentik/authentik/issues/19521
144
145        When using:
146        - Path-style addressing (bucket name goes in URL path, not subdomain)
147        - Custom domain that includes the bucket name (e.g., s3.example.com/bucket-name)
148
149        The bucket name should NOT appear twice in the final URL.
150
151        Example of the bug:
152        - custom_domain = "s3.example.com/authentik-media"
153        - boto3 presigned URL = "http://s3.example.com/authentik-media/media/public/file.png?..."
154        - Buggy result = "https://s3.example.com/authentik-media/authentik-media/media/public/file.png?..."
155        """
156        bucket_name = self.media_s3_bucket_name
157
158        # Custom domain includes the bucket name
159        custom_domain = f"localhost:8020/{bucket_name}"
160
161        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
162            url = self.media_s3_backend.file_url("application-icons/test.svg", use_cache=False)
163
164        # The bucket name should appear exactly once in the URL path, not twice
165        bucket_occurrences = url.count(bucket_name)
166        self.assertEqual(
167            bucket_occurrences,
168            1,
169            f"Bucket name '{bucket_name}' appears {bucket_occurrences} times in URL, expected 1. "
170            f"URL: {url}",
171        )

Test file_url doesn't duplicate bucket name when custom_domain includes bucket.

Regression test for https://github.com/goauthentik/authentik/issues/19521

When using:

  • Path-style addressing (bucket name goes in URL path, not subdomain)
  • Custom domain that includes the bucket name (e.g., s3.example.com/bucket-name)

The bucket name should NOT appear twice in the final URL.

Example of the bug:

  • custom_domain = "s3.example.com/authentik-media"
  • boto3 presigned URL = "http://s3.example.com/authentik-media/media/public/file.png?..."
  • Buggy result = "https://s3.example.com/authentik-media/authentik-media/media/public/file.png?..."
@CONFIG.patch('storage.s3.secure_urls', False)
@CONFIG.patch('storage.s3.addressing_style', 'path')
def test_file_url_custom_domain_resigns_for_custom_host(self): 
173    @CONFIG.patch("storage.s3.secure_urls", False)
174    @CONFIG.patch("storage.s3.addressing_style", "path")
175    def test_file_url_custom_domain_resigns_for_custom_host(self):
176        """Test presigned URLs are signed for the custom domain host.
177
178        Host-changing custom domains must produce a signature query string for
179        the public host, not reuse the internal endpoint signature.
180        """
181        bucket_name = self.media_s3_bucket_name
182        key_name = "application-icons/test.svg"
183        custom_domain = f"files.example.test:8020/{bucket_name}"
184
185        endpoint_signed_url = self.media_s3_backend.client.generate_presigned_url(
186            "get_object",
187            Params={
188                "Bucket": bucket_name,
189                "Key": f"{self.media_s3_backend.base_path}/{key_name}",
190            },
191            ExpiresIn=900,
192            HttpMethod="GET",
193        )
194
195        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
196            custom_url = self.media_s3_backend.file_url(key_name, use_cache=False)
197
198        endpoint_parts = urlsplit(endpoint_signed_url)
199        custom_parts = urlsplit(custom_url)
200
201        self.assertEqual(custom_parts.scheme, "http")
202        self.assertEqual(custom_parts.netloc, "files.example.test:8020")
203        self.assertEqual(parse_qs(custom_parts.query)["X-Amz-SignedHeaders"], ["host"])
204        self.assertNotEqual(
205            custom_parts.query,
206            endpoint_parts.query,
207            "Custom-domain URLs must be signed for the public host, not reuse the endpoint "
208            "signature query string.",
209        )

Test presigned URLs are signed for the custom domain host.

Host-changing custom domains must produce a signature query string for the public host, not reuse the internal endpoint signature.

def test_themed_urls_without_theme_variable(self): 
211    def test_themed_urls_without_theme_variable(self):
212        """Test themed_urls returns None when filename has no %(theme)s"""
213        result = self.media_s3_backend.themed_urls("logo.png")
214        self.assertIsNone(result)

Test themed_urls returns None when filename has no %(theme)s

def test_themed_urls_with_theme_variable(self): 
216    def test_themed_urls_with_theme_variable(self):
217        """Test themed_urls returns dict of presigned URLs for each theme"""
218        result = self.media_s3_backend.themed_urls("logo-%(theme)s.png")
219
220        self.assertIsInstance(result, dict)
221        self.assertIn("light", result)
222        self.assertIn("dark", result)
223
224        # Check URLs are valid presigned URLs with correct file paths
225        self.assertIn("logo-light.png", result["light"])
226        self.assertIn("logo-dark.png", result["dark"])
227        self.assertIn("X-Amz-Signature=", result["light"])
228        self.assertIn("X-Amz-Signature=", result["dark"])

Test themed_urls returns dict of presigned URLs for each theme

def test_themed_urls_multiple_theme_variables(self): 
230    def test_themed_urls_multiple_theme_variables(self):
231        """Test themed_urls with multiple %(theme)s in path"""
232        result = self.media_s3_backend.themed_urls("%(theme)s/logo-%(theme)s.svg")
233
234        self.assertIsInstance(result, dict)
235        self.assertIn("light/logo-light.svg", result["light"])
236        self.assertIn("dark/logo-dark.svg", result["dark"])

Test themed_urls with multiple %(theme)s in path

def test_save_file_sets_content_type_svg(self): 
238    def test_save_file_sets_content_type_svg(self):
239        """Test save_file sets correct ContentType for SVG files"""
240        self.media_s3_backend.save_file("test.svg", b"<svg></svg>")
241
242        response = self.media_s3_backend.client.head_object(
243            Bucket=self.media_s3_bucket_name,
244            Key="media/public/test.svg",
245        )
246        self.assertEqual(response["ContentType"], "image/svg+xml")

Test save_file sets correct ContentType for SVG files

def test_save_file_sets_content_type_png(self): 
248    def test_save_file_sets_content_type_png(self):
249        """Test save_file sets correct ContentType for PNG files"""
250        self.media_s3_backend.save_file("test.png", b"\x89PNG\r\n\x1a\n")
251
252        response = self.media_s3_backend.client.head_object(
253            Bucket=self.media_s3_bucket_name,
254            Key="media/public/test.png",
255        )
256        self.assertEqual(response["ContentType"], "image/png")

Test save_file sets correct ContentType for PNG files

def test_save_file_stream_sets_content_type(self): 
258    def test_save_file_stream_sets_content_type(self):
259        """Test save_file_stream sets correct ContentType"""
260        with self.media_s3_backend.save_file_stream("test.css") as f:
261            f.write(b"body { color: red; }")
262
263        response = self.media_s3_backend.client.head_object(
264            Bucket=self.media_s3_bucket_name,
265            Key="media/public/test.css",
266        )
267        self.assertEqual(response["ContentType"], "text/css")

Test save_file_stream sets correct ContentType

def test_save_file_unknown_extension_octet_stream(self): 
269    def test_save_file_unknown_extension_octet_stream(self):
270        """Test save_file sets octet-stream for unknown extensions"""
271        self.media_s3_backend.save_file("test.unknownext123", b"data")
272
273        response = self.media_s3_backend.client.head_object(
274            Bucket=self.media_s3_bucket_name,
275            Key="media/public/test.unknownext123",
276        )
277        self.assertEqual(response["ContentType"], "application/octet-stream")

Test save_file sets octet-stream for unknown extensions

Inherited Members
authentik.admin.files.tests.utils.FileTestS3BackendMixin
tearDown