authentik.admin.files.backends.base

  1import mimetypes
  2from collections.abc import Callable, Generator, Iterator
  3from typing import cast
  4
  5from django.core.cache import cache
  6from django.http.request import HttpRequest
  7from structlog.stdlib import get_logger
  8
  9from authentik.admin.files.usage import FileUsage
 10
 11CACHE_PREFIX = "goauthentik.io/admin/files"
 12LOGGER = get_logger()
 13
 14# Theme variable placeholder for theme-specific files like logo-%(theme)s.png
 15THEME_VARIABLE = "%(theme)s"
 16
 17
 18def get_content_type(name: str) -> str:
 19    """Get MIME type for a file based on its extension."""
 20    content_type, _ = mimetypes.guess_type(name)
 21    return content_type or "application/octet-stream"
 22
 23
 24def get_valid_themes() -> list[str]:
 25    """Get valid themes that can be substituted for %(theme)s."""
 26    from authentik.brands.api import Themes
 27
 28    return [t.value for t in Themes if t != Themes.AUTOMATIC]
 29
 30
 31def has_theme_variable(name: str) -> bool:
 32    """Check if filename contains %(theme)s variable."""
 33    return THEME_VARIABLE in name
 34
 35
 36def substitute_theme(name: str, theme: str) -> str:
 37    """Replace %(theme)s with the given theme."""
 38    return name.replace(THEME_VARIABLE, theme)
 39
 40
 41class Backend:
 42    """
 43    Base class for file storage backends.
 44
 45    Class attributes:
 46        allowed_usages: List of usages that can be used with this backend
 47    """
 48
 49    allowed_usages: list[FileUsage]
 50
 51    def __init__(self, usage: FileUsage):
 52        """
 53        Initialize backend for the given usage type.
 54
 55        Args:
 56            usage: FileUsage type enum value
 57        """
 58        self.usage = usage
 59        LOGGER.debug(
 60            "Initializing storage backend",
 61            backend=self.__class__.__name__,
 62            usage=usage.value,
 63        )
 64
 65    def supports_file(self, name: str) -> bool:
 66        """
 67        Check if this backend can handle the given file path.
 68
 69        Args:
 70            name: File path to check
 71
 72        Returns:
 73            True if this backend supports this file path
 74        """
 75        raise NotImplementedError
 76
 77    def list_files(self) -> Generator[str]:
 78        """
 79        List all files stored in this backend.
 80
 81        Yields:
 82            Relative file paths
 83        """
 84        raise NotImplementedError
 85
 86    def file_url(
 87        self,
 88        name: str,
 89        request: HttpRequest | None = None,
 90        use_cache: bool = True,
 91    ) -> str:
 92        """
 93        Get URL for accessing the file.
 94
 95        Args:
 96            file_path: Relative file path
 97            request: Optional Django HttpRequest for fully qualified URL building
 98            use_cache: whether to retrieve the URL from cache
 99
100        Returns:
101            URL to access the file (may be relative or absolute depending on backend)
102        """
103        raise NotImplementedError
104
105    def themed_urls(
106        self,
107        name: str,
108        request: HttpRequest | None = None,
109    ) -> dict[str, str] | None:
110        """
111        Get URLs for each theme variant when filename contains %(theme)s.
112
113        Args:
114            name: File path potentially containing %(theme)s
115            request: Optional Django HttpRequest for URL building
116
117        Returns:
118            Dict mapping theme to URL if %(theme)s present, None otherwise
119        """
120        if not has_theme_variable(name):
121            return None
122
123        return {
124            theme: self.file_url(substitute_theme(name, theme), request, use_cache=True)
125            for theme in get_valid_themes()
126        }
127
128
129class ManageableBackend(Backend):
130    """
131    Base class for manageable file storage backends.
132
133    Class attributes:
134        name: Canonical name of the storage backend, for use in configuration.
135    """
136
137    name: str
138
139    @property
140    def manageable(self) -> bool:
141        """
142        Whether this backend can actually be used for management.
143
144        Used only for management check, not for created the backend
145        """
146        raise NotImplementedError
147
148    def save_file(self, name: str, content: bytes) -> None:
149        """
150        Save file content to storage.
151
152        Args:
153            file_path: Relative file path
154            content: File content as bytes
155        """
156        raise NotImplementedError
157
158    def save_file_stream(self, name: str) -> Iterator:
159        """
160        Context manager for streaming file writes.
161
162        Args:
163            file_path: Relative file path
164
165        Returns:
166            Context manager that yields a writable file-like object
167
168        FileUsage:
169            with backend.save_file_stream("output.csv") as f:
170                f.write(b"data...")
171        """
172        raise NotImplementedError
173
174    def delete_file(self, name: str) -> None:
175        """
176        Delete file from storage.
177
178        Args:
179            file_path: Relative file path
180        """
181        raise NotImplementedError
182
183    def file_exists(self, name: str) -> bool:
184        """
185        Check if a file exists.
186
187        Args:
188            file_path: Relative file path
189
190        Returns:
191            True if file exists, False otherwise
192        """
193        raise NotImplementedError
194
195    def _cache_get_or_set(
196        self,
197        name: str,
198        request: HttpRequest | None,
199        default: Callable[[str, HttpRequest | None], str],
200        timeout: int,
201    ) -> str:
202        timeout_ignore = 60
203        timeout = int(timeout * 0.67)
204        if timeout < timeout_ignore:
205            timeout = 0
206
207        request_key = "None"
208        if request is not None:
209            request_key = f"{request.build_absolute_uri('/')}"
210        cache_key = f"{CACHE_PREFIX}/{self.name}/{self.usage}/{request_key}/{name}"
211
212        return cast(str, cache.get_or_set(cache_key, lambda: default(name, request), timeout))