authentik.admin.files.backends.file

  1import os
  2from collections.abc import Generator, Iterator
  3from contextlib import contextmanager
  4from datetime import timedelta
  5from hashlib import sha256
  6from pathlib import Path
  7
  8import jwt
  9from django.conf import settings
 10from django.db import connection
 11from django.http.request import HttpRequest
 12from django.utils.timezone import now
 13
 14from authentik.admin.files.backends.base import ManageableBackend
 15from authentik.admin.files.usage import FileUsage
 16from authentik.lib.config import CONFIG
 17from authentik.lib.utils.time import timedelta_from_string
 18
 19
 20class FileBackend(ManageableBackend):
 21    """Local filesystem backend for file storage.
 22
 23    Stores files in a local directory structure:
 24    - Path: {base_dir}/{usage}/{schema}/{filename}
 25    - Supports full file management (upload, delete, list)
 26    - Used when storage.backend=file (default)
 27    """
 28
 29    name = "file"
 30    allowed_usages = list(FileUsage)  # All usages
 31
 32    @property
 33    def _base_dir(self) -> Path:
 34        return Path(
 35            CONFIG.get(
 36                f"storage.{self.usage.value}.{self.name}.path",
 37                CONFIG.get(f"storage.{self.name}.path", "./data"),
 38            )
 39        )
 40
 41    @property
 42    def base_path(self) -> Path:
 43        """Path structure: {base_dir}/{usage}/{schema}"""
 44        return self._base_dir / self.usage.value / connection.schema_name
 45
 46    @property
 47    def manageable(self) -> bool:
 48        # Check _base_dir (the mount point, e.g. /data) rather than base_path
 49        # (which includes usage/schema subdirs, e.g. /data/media/public).
 50        # The subdirectories are created on first file write via mkdir(parents=True)
 51        # in save_file(), so requiring them to exist beforehand would prevent
 52        # file creation on fresh installs.
 53        return (
 54            self._base_dir.exists()
 55            and (self._base_dir.is_mount() or (self._base_dir / self.usage.value).is_mount())
 56            or (settings.DEBUG or settings.TEST)
 57        )
 58
 59    def supports_file(self, name: str) -> bool:
 60        """We support all files"""
 61        return True
 62
 63    def list_files(self) -> Generator[str]:
 64        """List all files returning relative paths from base_path."""
 65        for root, _, files in os.walk(self.base_path):
 66            for file in files:
 67                full_path = Path(root) / file
 68                rel_path = full_path.relative_to(self.base_path)
 69                yield str(rel_path)
 70
 71    def file_url(
 72        self,
 73        name: str,
 74        request: HttpRequest | None = None,
 75        use_cache: bool = True,
 76    ) -> str:
 77        """Get URL for accessing the file."""
 78        expires_in = timedelta_from_string(
 79            CONFIG.get(
 80                f"storage.{self.usage.value}.{self.name}.url_expiry",
 81                CONFIG.get(f"storage.{self.name}.url_expiry", "minutes=15"),
 82            )
 83        )
 84
 85        def _file_url(name: str, request: HttpRequest | None) -> str:
 86            prefix = CONFIG.get("web.path", "/")[:-1]
 87            path = f"{self.usage.value}/{connection.schema_name}/{name}"
 88            token = jwt.encode(
 89                payload={
 90                    "path": path,
 91                    "exp": now() + expires_in,
 92                    "nbf": now() - timedelta(seconds=15),
 93                },
 94                key=sha256(f"{settings.SECRET_KEY}:{self.usage}".encode()).hexdigest(),
 95                algorithm="HS256",
 96            )
 97            url = f"{prefix}/files/{path}?token={token}"
 98            if request is None:
 99                return url
100            return request.build_absolute_uri(url)
101
102        if use_cache:
103            timeout = int(expires_in.total_seconds())
104            return self._cache_get_or_set(name, request, _file_url, timeout)
105        else:
106            return _file_url(name, request)
107
108    def save_file(self, name: str, content: bytes) -> None:
109        """Save file to local filesystem."""
110        path = self.base_path / Path(name)
111        path.parent.mkdir(parents=True, exist_ok=True)
112        with open(path, "w+b") as f:
113            f.write(content)
114
115    @contextmanager
116    def save_file_stream(self, name: str) -> Iterator:
117        """Context manager for streaming file writes to local filesystem."""
118        path = self.base_path / Path(name)
119        path.parent.mkdir(parents=True, exist_ok=True)
120        with open(path, "wb") as f:
121            yield f
122
123    def delete_file(self, name: str) -> None:
124        """Delete file from local filesystem."""
125        path = self.base_path / Path(name)
126        path.unlink(missing_ok=True)
127
128    def file_exists(self, name: str) -> bool:
129        """Check if a file exists."""
130        path = self.base_path / Path(name)
131        return path.exists()