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        use_cache: bool = True,
110    ) -> dict[str, str] | None:
111        """
112        Get URLs for each theme variant when filename contains %(theme)s.
113
114        Args:
115            name: File path potentially containing %(theme)s
116            request: Optional Django HttpRequest for URL building
117
118        Returns:
119            Dict mapping theme to URL if %(theme)s present, None otherwise
120        """
121        if not has_theme_variable(name):
122            return None
123
124        return {
125            theme: self.file_url(substitute_theme(name, theme), request, use_cache=use_cache)
126            for theme in get_valid_themes()
127        }
128
129
130class ManageableBackend(Backend):
131    """
132    Base class for manageable file storage backends.
133
134    Class attributes:
135        name: Canonical name of the storage backend, for use in configuration.
136    """
137
138    name: str
139
140    @property
141    def manageable(self) -> bool:
142        """
143        Whether this backend can actually be used for management.
144
145        Used only for management check, not for created the backend
146        """
147        raise NotImplementedError
148
149    def save_file(self, name: str, content: bytes) -> None:
150        """
151        Save file content to storage.
152
153        Args:
154            file_path: Relative file path
155            content: File content as bytes
156        """
157        raise NotImplementedError
158
159    def save_file_stream(self, name: str) -> Iterator:
160        """
161        Context manager for streaming file writes.
162
163        Args:
164            file_path: Relative file path
165
166        Returns:
167            Context manager that yields a writable file-like object
168
169        FileUsage:
170            with backend.save_file_stream("output.csv") as f:
171                f.write(b"data...")
172        """
173        raise NotImplementedError
174
175    def delete_file(self, name: str) -> None:
176        """
177        Delete file from storage.
178
179        Args:
180            file_path: Relative file path
181        """
182        raise NotImplementedError
183
184    def file_exists(self, name: str) -> bool:
185        """
186        Check if a file exists.
187
188        Args:
189            file_path: Relative file path
190
191        Returns:
192            True if file exists, False otherwise
193        """
194        raise NotImplementedError
195
196    def _cache_get_or_set(
197        self,
198        name: str,
199        request: HttpRequest | None,
200        default: Callable[[str, HttpRequest | None], str],
201        timeout: int,
202    ) -> str:
203        timeout_ignore = 60
204        timeout = int(timeout * 0.67)
205        if timeout < timeout_ignore:
206            timeout = 0
207
208        request_key = "None"
209        if request is not None:
210            request_key = f"{request.build_absolute_uri('/')}"
211        cache_key = f"{CACHE_PREFIX}/{self.name}/{self.usage}/{request_key}/{name}"
212
213        return cast(str, cache.get_or_set(cache_key, lambda: default(name, request), timeout))