authentik.admin.files.manager

  1from collections.abc import Generator, Iterator
  2
  3from django.core.exceptions import ImproperlyConfigured
  4from django.http.request import HttpRequest
  5from rest_framework.request import Request
  6from structlog.stdlib import get_logger
  7
  8from authentik.admin.files.backends.base import ManageableBackend
  9from authentik.admin.files.backends.file import FileBackend
 10from authentik.admin.files.backends.passthrough import PassthroughBackend
 11from authentik.admin.files.backends.s3 import S3Backend
 12from authentik.admin.files.backends.static import StaticBackend
 13from authentik.admin.files.usage import FileUsage
 14from authentik.lib.config import CONFIG
 15
 16LOGGER = get_logger()
 17
 18
 19_FILE_BACKENDS = [
 20    StaticBackend,
 21    PassthroughBackend,
 22    FileBackend,
 23    S3Backend,
 24]
 25
 26
 27class FileManager:
 28    def __init__(self, usage: FileUsage) -> None:
 29        management_backend_name = CONFIG.get(
 30            f"storage.{usage.value}.backend",
 31            CONFIG.get("storage.backend", "file"),
 32        )
 33
 34        self.management_backend = None
 35        for backend in _FILE_BACKENDS:
 36            if issubclass(backend, ManageableBackend) and backend.name == management_backend_name:
 37                self.management_backend = backend(usage)
 38        if self.management_backend is None:
 39            LOGGER.warning(
 40                f"Storage backend configuration for {usage.value} is "
 41                f"invalid: {management_backend_name}"
 42            )
 43
 44        self.backends = []
 45        for backend in _FILE_BACKENDS:
 46            if usage not in backend.allowed_usages:
 47                continue
 48            if isinstance(self.management_backend, backend):
 49                self.backends.append(self.management_backend)
 50            elif not issubclass(backend, ManageableBackend):
 51                self.backends.append(backend(usage))
 52
 53    @property
 54    def manageable(self) -> bool:
 55        """
 56        Whether this file manager is able to manage files.
 57        """
 58        return self.management_backend is not None and self.management_backend.manageable
 59
 60    def list_files(self, manageable_only: bool = False) -> Generator[str]:
 61        """
 62        List available files.
 63        """
 64        for backend in self.backends:
 65            if manageable_only and not isinstance(backend, ManageableBackend):
 66                continue
 67            yield from backend.list_files()
 68
 69    def file_url(
 70        self,
 71        name: str | None,
 72        request: HttpRequest | Request | None = None,
 73        use_cache: bool = True,
 74    ) -> str:
 75        """
 76        Get URL for accessing the file.
 77
 78        Set ``use_cache=False`` when the caller needs a fresh signed URL instead
 79        of a cached one, for example when serializing flow/login payloads that
 80        may be refreshed after the previous JWT has expired.
 81        """
 82        if not name:
 83            return ""
 84
 85        if isinstance(request, Request):
 86            request = request._request
 87
 88        for backend in self.backends:
 89            if backend.supports_file(name):
 90                return backend.file_url(name, request, use_cache=use_cache)
 91
 92        LOGGER.warning(f"Could not find file backend for file: {name}")
 93        return ""
 94
 95    def themed_urls(
 96        self,
 97        name: str | None,
 98        request: HttpRequest | Request | None = None,
 99        use_cache: bool = True,
100    ) -> dict[str, str] | None:
101        """
102        Get URLs for each theme variant when filename contains %(theme)s.
103
104        ``use_cache`` has the same semantics as ``file_url()`` and allows
105        callers to force regeneration of expiring signed URLs.
106
107        Returns dict mapping theme to URL if %(theme)s present, None otherwise.
108        """
109        if not name:
110            return None
111
112        if isinstance(request, Request):
113            request = request._request
114
115        for backend in self.backends:
116            if backend.supports_file(name):
117                return backend.themed_urls(name, request, use_cache=use_cache)
118
119        return None
120
121    def _check_manageable(self) -> None:
122        if not self.manageable:
123            raise ImproperlyConfigured("No file management backend configured.")
124
125    def save_file(self, file_path: str, content: bytes) -> None:
126        """
127        Save file contents to storage.
128        """
129        self._check_manageable()
130        assert self.management_backend is not None  # nosec
131        return self.management_backend.save_file(file_path, content)
132
133    def save_file_stream(self, file_path: str) -> Iterator:
134        """
135        Context manager for streaming file writes.
136
137        Args:
138            file_path: Relative file path
139
140        Returns:
141            Context manager that yields a writable file-like object
142
143        Usage:
144            with manager.save_file_stream("output.csv") as f:
145                f.write(b"data...")
146        """
147        self._check_manageable()
148        assert self.management_backend is not None  # nosec
149        return self.management_backend.save_file_stream(file_path)
150
151    def delete_file(self, file_path: str) -> None:
152        """
153        Delete file from storage.
154        """
155        self._check_manageable()
156        assert self.management_backend is not None  # nosec
157        return self.management_backend.delete_file(file_path)
158
159    def file_exists(self, file_path: str) -> bool:
160        """
161        Check if a file exists.
162        """
163        self._check_manageable()
164        assert self.management_backend is not None  # nosec
165        return self.management_backend.file_exists(file_path)
166
167
168MANAGERS = {usage: FileManager(usage) for usage in list(FileUsage)}
169
170
171def get_file_manager(usage: FileUsage) -> FileManager:
172    return MANAGERS[usage]