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        if not name:
 79            return ""
 80
 81        if isinstance(request, Request):
 82            request = request._request
 83
 84        for backend in self.backends:
 85            if backend.supports_file(name):
 86                return backend.file_url(name, request)
 87
 88        LOGGER.warning(f"Could not find file backend for file: {name}")
 89        return ""
 90
 91    def themed_urls(
 92        self,
 93        name: str | None,
 94        request: HttpRequest | Request | None = None,
 95    ) -> dict[str, str] | None:
 96        """
 97        Get URLs for each theme variant when filename contains %(theme)s.
 98
 99        Returns dict mapping theme to URL if %(theme)s present, None otherwise.
100        """
101        if not name:
102            return None
103
104        if isinstance(request, Request):
105            request = request._request
106
107        for backend in self.backends:
108            if backend.supports_file(name):
109                return backend.themed_urls(name, request)
110
111        return None
112
113    def _check_manageable(self) -> None:
114        if not self.manageable:
115            raise ImproperlyConfigured("No file management backend configured.")
116
117    def save_file(self, file_path: str, content: bytes) -> None:
118        """
119        Save file contents to storage.
120        """
121        self._check_manageable()
122        assert self.management_backend is not None  # nosec
123        return self.management_backend.save_file(file_path, content)
124
125    def save_file_stream(self, file_path: str) -> Iterator:
126        """
127        Context manager for streaming file writes.
128
129        Args:
130            file_path: Relative file path
131
132        Returns:
133            Context manager that yields a writable file-like object
134
135        Usage:
136            with manager.save_file_stream("output.csv") as f:
137                f.write(b"data...")
138        """
139        self._check_manageable()
140        assert self.management_backend is not None  # nosec
141        return self.management_backend.save_file_stream(file_path)
142
143    def delete_file(self, file_path: str) -> None:
144        """
145        Delete file from storage.
146        """
147        self._check_manageable()
148        assert self.management_backend is not None  # nosec
149        return self.management_backend.delete_file(file_path)
150
151    def file_exists(self, file_path: str) -> bool:
152        """
153        Check if a file exists.
154        """
155        self._check_manageable()
156        assert self.management_backend is not None  # nosec
157        return self.management_backend.file_exists(file_path)
158
159
160MANAGERS = {usage: FileManager(usage) for usage in list(FileUsage)}
161
162
163def get_file_manager(usage: FileUsage) -> FileManager:
164    return MANAGERS[usage]