mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-14 18:38:11 +02:00
Initial commit
This commit is contained in:
commit
8aa223bc7b
40 files changed
+4320
No files matched your search
@@ -0,0 +1,53 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import Depends, HTTPException, Request
|
||||
|
||||
from Users import GetUser
|
||||
|
||||
|
||||
# In-memory only for now: sessions don't survive a container restart, so
|
||||
# everyone just logs in again. Fine for a handful of trusted tailnet users;
|
||||
# worth revisiting if that stops being true.
|
||||
ActiveSessions: dict[str, str] = {}
|
||||
|
||||
SessionCookieName = "SaraViewSession"
|
||||
|
||||
|
||||
def CreateSession(Username: str) -> str:
|
||||
Token = secrets.token_urlsafe(32)
|
||||
ActiveSessions[Token] = Username
|
||||
return Token
|
||||
|
||||
|
||||
def DeleteSession(Token: str) -> None:
|
||||
ActiveSessions.pop(Token, None)
|
||||
|
||||
|
||||
def GetCurrentUser(request: Request) -> dict[str, Any]:
|
||||
Token = request.cookies.get(SessionCookieName)
|
||||
Username = ActiveSessions.get(Token) if Token else None
|
||||
|
||||
if Username is None:
|
||||
raise HTTPException(status_code=401, detail="Not logged in.")
|
||||
|
||||
User = GetUser(Username)
|
||||
|
||||
if User is None:
|
||||
raise HTTPException(status_code=401, detail="Session user no longer exists.")
|
||||
|
||||
return {
|
||||
"Username": Username,
|
||||
"IsAdmin": User.get("IsAdmin", False),
|
||||
"Colour": User.get("Colour", "Purple"),
|
||||
}
|
||||
|
||||
|
||||
def RequireAdmin(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
if not CurrentUser.get("IsAdmin"):
|
||||
raise HTTPException(status_code=403, detail="Admin access required.")
|
||||
|
||||
return CurrentUser
|
||||
@@ -0,0 +1,75 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import difflib
|
||||
import tomllib
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
|
||||
from Settings import GetDataDirectory
|
||||
|
||||
|
||||
def GetBlameDirectory() -> Path:
|
||||
return GetDataDirectory() / "Blame"
|
||||
|
||||
|
||||
def GetBlamePath(RootName: str, RelativePath: str) -> Path:
|
||||
# Mirrors the file's own relative path under the Blame/ folder, keyed by
|
||||
# root so files with the same relative path in different roots don't
|
||||
# collide. Validity of RelativePath itself is checked by the caller via
|
||||
# Core.ResolvePath before this is ever used.
|
||||
return GetBlameDirectory() / RootName / f"{RelativePath}.blame.toml"
|
||||
|
||||
|
||||
def LoadBlame(RootName: str, RelativePath: str) -> dict[str, dict[str, Any]]:
|
||||
BlamePath = GetBlamePath(RootName, RelativePath)
|
||||
|
||||
if not BlamePath.is_file():
|
||||
return {}
|
||||
|
||||
with BlamePath.open("rb") as BlameFile:
|
||||
Loaded = tomllib.load(BlameFile)
|
||||
|
||||
return Loaded.get("Lines", {})
|
||||
|
||||
|
||||
def SaveBlame(RootName: str, RelativePath: str, Blame: dict[str, dict[str, Any]]) -> None:
|
||||
BlamePath = GetBlamePath(RootName, RelativePath)
|
||||
BlamePath.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with BlamePath.open("wb") as BlameFile:
|
||||
tomli_w.dump({"Lines": Blame}, BlameFile)
|
||||
|
||||
|
||||
def UpdateBlameOnSave(
|
||||
RootName: str, RelativePath: str, OldContent: str, NewContent: str, Username: str
|
||||
) -> None:
|
||||
"""Recompute line attribution after a save. Unchanged lines (matched via
|
||||
the same diff algorithm git's own blame uses) keep their prior author and
|
||||
timestamp; changed or new lines are stamped with the current user & now.
|
||||
"""
|
||||
OldLines = OldContent.splitlines()
|
||||
NewLines = NewContent.splitlines()
|
||||
PreviousBlame = LoadBlame(RootName, RelativePath)
|
||||
Timestamp = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
Matcher = difflib.SequenceMatcher(None, OldLines, NewLines, autojunk=False)
|
||||
NewBlame: dict[str, dict[str, Any]] = {}
|
||||
|
||||
for Tag, OldStart, OldEnd, NewStart, NewEnd in Matcher.get_opcodes():
|
||||
if Tag == "equal":
|
||||
for Offset in range(OldEnd - OldStart):
|
||||
OldLineNumber = str(OldStart + Offset + 1)
|
||||
NewLineNumber = str(NewStart + Offset + 1)
|
||||
NewBlame[NewLineNumber] = PreviousBlame.get(
|
||||
OldLineNumber, {"Username": Username, "Timestamp": Timestamp}
|
||||
)
|
||||
else:
|
||||
for Offset in range(NewEnd - NewStart):
|
||||
NewLineNumber = str(NewStart + Offset + 1)
|
||||
NewBlame[NewLineNumber] = {"Username": Username, "Timestamp": Timestamp}
|
||||
|
||||
SaveBlame(RootName, RelativePath, NewBlame)
|
||||
@@ -0,0 +1,85 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import tomllib
|
||||
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from typing import Any
|
||||
|
||||
from Core import Config
|
||||
|
||||
|
||||
CachedSaraTagCore: ModuleType | None = None
|
||||
CachedSaraTagCoreLoaded = False
|
||||
|
||||
|
||||
def GetSaraTagCore() -> ModuleType | None:
|
||||
"""Load SaraTag's Core.py directly by file path, under a private module
|
||||
name, so it never collides with SaraView's own Core module in sys.modules.
|
||||
Cached after the first successful load.
|
||||
"""
|
||||
global CachedSaraTagCore, CachedSaraTagCoreLoaded
|
||||
|
||||
if CachedSaraTagCoreLoaded:
|
||||
return CachedSaraTagCore
|
||||
|
||||
CachedSaraTagCoreLoaded = True
|
||||
|
||||
SaraTagConfig = Config.get("SaraTag", {})
|
||||
|
||||
if not SaraTagConfig.get("Enabled", False):
|
||||
return None
|
||||
|
||||
CoreModulePath = SaraTagConfig.get("CoreModulePath")
|
||||
|
||||
if not CoreModulePath:
|
||||
return None
|
||||
|
||||
CoreFilePath = Path(CoreModulePath).expanduser().resolve() / "Core.py"
|
||||
|
||||
if not CoreFilePath.is_file():
|
||||
return None
|
||||
|
||||
ModuleSpec = importlib.util.spec_from_file_location("SaraTagCore", CoreFilePath)
|
||||
|
||||
if ModuleSpec is None or ModuleSpec.loader is None:
|
||||
return None
|
||||
|
||||
SaraTagCoreModule = importlib.util.module_from_spec(ModuleSpec)
|
||||
ModuleSpec.loader.exec_module(SaraTagCoreModule)
|
||||
|
||||
CachedSaraTagCore = SaraTagCoreModule
|
||||
return CachedSaraTagCore
|
||||
|
||||
|
||||
def ReadBookForFile(FilePath: Path) -> dict[str, Any] | None:
|
||||
"""Return a file's resolved Book (tag values + Lexicon definitions), or
|
||||
None if SaraTag integration is disabled or the file has no Book.
|
||||
"""
|
||||
SaraTagCore = GetSaraTagCore()
|
||||
|
||||
if SaraTagCore is None:
|
||||
return None
|
||||
|
||||
BookPath = SaraTagCore.GetBookPath(FilePath)
|
||||
|
||||
if not BookPath.is_file():
|
||||
return None
|
||||
|
||||
with BookPath.open("rb") as BookFile:
|
||||
Metadata = tomllib.load(BookFile)
|
||||
|
||||
ParentValue = Metadata.get("Parent")
|
||||
Definitions: dict[str, Any] = {}
|
||||
|
||||
if isinstance(ParentValue, str):
|
||||
try:
|
||||
LexiconPath = SaraTagCore.ResolveBookLexiconPath(FilePath.parent, ParentValue)
|
||||
Definitions = SaraTagCore.ResolveLexiconTags(LexiconPath)
|
||||
except Exception:
|
||||
Definitions = {}
|
||||
|
||||
Tags = {Key: Value for Key, Value in Metadata.items() if Key != "Parent"}
|
||||
|
||||
return {"Parent": ParentValue, "Tags": Tags, "Definitions": Definitions}
|
||||
@@ -0,0 +1,67 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import tomllib
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
ApplicationDirectory = Path(__file__).resolve().parent
|
||||
ConfigPath = ApplicationDirectory / "SaraViewConf.toml"
|
||||
|
||||
|
||||
def LoadConfig() -> dict[str, Any]:
|
||||
"""Load SaraView's TOML configuration from disk."""
|
||||
if not ConfigPath.is_file():
|
||||
raise FileNotFoundError(f"Missing configuration file: {ConfigPath}")
|
||||
|
||||
with ConfigPath.open("rb") as ConfigFile:
|
||||
return tomllib.load(ConfigFile)
|
||||
|
||||
|
||||
Config = LoadConfig()
|
||||
|
||||
|
||||
def GetRoots() -> dict[str, Path]:
|
||||
"""Return the configured named roots as resolved, absolute paths."""
|
||||
RawRoots = Config.get("Roots", {})
|
||||
Roots: dict[str, Path] = {}
|
||||
|
||||
for Name, RawPath in RawRoots.items():
|
||||
Roots[Name] = Path(RawPath).expanduser().resolve()
|
||||
|
||||
return Roots
|
||||
|
||||
|
||||
def ResolvePath(RootName: str, RelativePath: str) -> Path:
|
||||
"""Resolve a client-supplied relative path against a named root.
|
||||
|
||||
Raises ValueError if the root is unknown or the resolved path escapes it.
|
||||
"""
|
||||
Roots = GetRoots()
|
||||
Root = Roots.get(RootName)
|
||||
|
||||
if Root is None:
|
||||
raise ValueError(f"Unknown root: {RootName}")
|
||||
|
||||
Candidate = (Root / RelativePath.lstrip("/")).resolve()
|
||||
|
||||
if Candidate != Root and Root not in Candidate.parents:
|
||||
raise ValueError("Path escapes the configured root.")
|
||||
|
||||
return Candidate
|
||||
|
||||
|
||||
def IsInternalPath(EntryPath: Path) -> bool:
|
||||
"""Return True if an entry should be hidden as a SaraTag internal file."""
|
||||
SaraTagConfig = Config.get("SaraTag", {})
|
||||
|
||||
if not SaraTagConfig.get("HideInternalFiles", True):
|
||||
return False
|
||||
|
||||
Patterns = SaraTagConfig.get(
|
||||
"InternalPatterns", [".saratag", ".SaraTag*", "*.book.toml"]
|
||||
)
|
||||
|
||||
return any(fnmatch.fnmatch(EntryPath.name, Pattern) for Pattern in Patterns)
|
||||
@@ -0,0 +1,15 @@
|
||||
[Palette]
|
||||
Background = "#1e1e2e"
|
||||
Surface = "#181825"
|
||||
SurfaceAlt = "#313244"
|
||||
Border = "#45475a"
|
||||
Text = "#cdd6f4"
|
||||
TextMuted = "#6c7086"
|
||||
Black = "#11111b"
|
||||
White = "#cdd6f4"
|
||||
Red = "#f38ba8"
|
||||
Orange = "#fab387"
|
||||
Yellow = "#f9e2af"
|
||||
Green = "#a6e3a1"
|
||||
Blue = "#89b4fa"
|
||||
Purple = "#cba6f7"
|
||||
@@ -0,0 +1,15 @@
|
||||
[Palette]
|
||||
Background = "#1b2420"
|
||||
Surface = "#141b17"
|
||||
SurfaceAlt = "#243029"
|
||||
Border = "#334539"
|
||||
Text = "#d6e5d0"
|
||||
TextMuted = "#7c9184"
|
||||
Black = "#0d120f"
|
||||
White = "#eef5ea"
|
||||
Red = "#e0685a"
|
||||
Orange = "#e0a458"
|
||||
Yellow = "#e0c458"
|
||||
Green = "#7fb069"
|
||||
Blue = "#5c9e8f"
|
||||
Purple = "#9b7fb0"
|
||||
@@ -0,0 +1,15 @@
|
||||
[Palette]
|
||||
Background = "#282828"
|
||||
Surface = "#1d2021"
|
||||
SurfaceAlt = "#3c3836"
|
||||
Border = "#504945"
|
||||
Text = "#ebdbb2"
|
||||
TextMuted = "#a89984"
|
||||
Black = "#1d2021"
|
||||
White = "#fbf1c7"
|
||||
Red = "#fb4934"
|
||||
Orange = "#fe8019"
|
||||
Yellow = "#fabd2f"
|
||||
Green = "#b8bb26"
|
||||
Blue = "#83a598"
|
||||
Purple = "#d3869b"
|
||||
@@ -0,0 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def WriteTextFile(FilePath: Path, Content: str) -> None:
|
||||
"""Overwrite a text file's contents on disk."""
|
||||
FilePath.write_text(Content, encoding="utf-8")
|
||||
@@ -0,0 +1,130 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from Core import GetRoots, IsInternalPath, ResolvePath
|
||||
|
||||
|
||||
FileTypeMap = {
|
||||
".md": "markdown",
|
||||
".markdown": "markdown",
|
||||
".txt": "text",
|
||||
".toml": "config",
|
||||
".yaml": "config",
|
||||
".yml": "config",
|
||||
".json": "config",
|
||||
".cfg": "config",
|
||||
".ini": "config",
|
||||
".html": "html",
|
||||
".htm": "html",
|
||||
".png": "image",
|
||||
".jpg": "image",
|
||||
".jpeg": "image",
|
||||
".gif": "image",
|
||||
".webp": "image",
|
||||
".svg": "image",
|
||||
".mp4": "video",
|
||||
".webm": "video",
|
||||
".mov": "video",
|
||||
".mkv": "video",
|
||||
}
|
||||
|
||||
|
||||
def ClassifyFile(EntryPath: Path) -> str:
|
||||
"""Return a coarse file-type label used to pick a preview renderer."""
|
||||
return FileTypeMap.get(EntryPath.suffix.lower(), "unknown")
|
||||
|
||||
|
||||
def ListRoots() -> list[dict[str, Any]]:
|
||||
"""Return the configured roots as a flat list for the frontend."""
|
||||
return [{"Name": Name} for Name in GetRoots()]
|
||||
|
||||
|
||||
def CreateEntry(RootName: str, RelativePath: str, IsDirectory: bool) -> None:
|
||||
"""Create a new empty file or directory at the given path."""
|
||||
EntryPath = ResolvePath(RootName, RelativePath)
|
||||
|
||||
if EntryPath.exists():
|
||||
raise FileExistsError(f"{EntryPath.name} already exists.")
|
||||
|
||||
if IsDirectory:
|
||||
EntryPath.mkdir(parents=False)
|
||||
else:
|
||||
EntryPath.parent.mkdir(parents=True, exist_ok=True)
|
||||
EntryPath.touch()
|
||||
|
||||
|
||||
def MoveEntry(RootName: str, SourceRelativePath: str, DestinationRelativePath: str) -> None:
|
||||
"""Rename or move a file/directory within the same root. Used for both
|
||||
plain renames (same parent, new name) and full moves.
|
||||
"""
|
||||
SourcePath = ResolvePath(RootName, SourceRelativePath)
|
||||
DestinationPath = ResolvePath(RootName, DestinationRelativePath)
|
||||
|
||||
if not SourcePath.exists():
|
||||
raise FileNotFoundError(f"{SourcePath.name} does not exist.")
|
||||
|
||||
if DestinationPath.exists():
|
||||
raise FileExistsError(f"{DestinationPath.name} already exists.")
|
||||
|
||||
DestinationPath.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
SourcePath.rename(DestinationPath)
|
||||
except OSError:
|
||||
# Cross-device or other rename failures — fall back to a copy+delete.
|
||||
shutil.move(str(SourcePath), str(DestinationPath))
|
||||
|
||||
|
||||
def DeleteEntry(RootName: str, RelativePath: str) -> None:
|
||||
"""Delete a file, or an entire directory tree."""
|
||||
EntryPath = ResolvePath(RootName, RelativePath)
|
||||
|
||||
if not EntryPath.exists():
|
||||
raise FileNotFoundError(f"{EntryPath.name} does not exist.")
|
||||
|
||||
if EntryPath.is_dir():
|
||||
shutil.rmtree(EntryPath)
|
||||
else:
|
||||
EntryPath.unlink()
|
||||
|
||||
|
||||
def ListDirectory(RootName: str, RelativePath: str, ShowHidden: bool = False) -> list[dict[str, Any]]:
|
||||
"""List the immediate contents of a directory.
|
||||
|
||||
SaraTag's internal files are always hidden, regardless of ShowHidden.
|
||||
Dotfiles/dotfolders are hidden unless ShowHidden is True.
|
||||
"""
|
||||
DirectoryPath = ResolvePath(RootName, RelativePath)
|
||||
|
||||
if not DirectoryPath.is_dir():
|
||||
raise NotADirectoryError(f"{DirectoryPath} is not a directory.")
|
||||
|
||||
Entries: list[dict[str, Any]] = []
|
||||
|
||||
SortedChildren = sorted(
|
||||
DirectoryPath.iterdir(),
|
||||
key=lambda EntryPath: (EntryPath.is_file(), EntryPath.name.casefold()),
|
||||
)
|
||||
|
||||
for EntryPath in SortedChildren:
|
||||
if IsInternalPath(EntryPath):
|
||||
continue
|
||||
|
||||
if not ShowHidden and EntryPath.name.startswith("."):
|
||||
continue
|
||||
|
||||
IsDirectory = EntryPath.is_dir()
|
||||
Entries.append(
|
||||
{
|
||||
"Name": EntryPath.name,
|
||||
"IsDirectory": IsDirectory,
|
||||
"FileType": None if IsDirectory else ClassifyFile(EntryPath),
|
||||
"Size": None if IsDirectory else EntryPath.stat().st_size,
|
||||
}
|
||||
)
|
||||
|
||||
return Entries
|
||||
+459
@@ -0,0 +1,459 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import mimetypes
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ApplicationDirectory = Path(__file__).resolve().parent
|
||||
if str(ApplicationDirectory) not in sys.path:
|
||||
sys.path.insert(0, str(ApplicationDirectory))
|
||||
|
||||
from fastapi import Depends, FastAPI, File, HTTPException, Request, Response, UploadFile
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import PlainTextResponse, StreamingResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from Auth import CreateSession, DeleteSession, GetCurrentUser, RequireAdmin, SessionCookieName
|
||||
from BookReader import ReadBookForFile
|
||||
from Blame import LoadBlame, UpdateBlameOnSave
|
||||
from Core import ResolvePath
|
||||
from Editor import WriteTextFile
|
||||
from Explorer import ClassifyFile, CreateEntry, DeleteEntry, ListDirectory, ListRoots, MoveEntry
|
||||
from Preview import GetFileSize, GetMimeType, ParseRangeHeader, ReadTextFile, TextFileTypes
|
||||
from Settings import (
|
||||
DeleteTheme,
|
||||
ListThemeNames,
|
||||
LoadSettings,
|
||||
LoadTheme,
|
||||
PaletteRoles,
|
||||
SaveSettings,
|
||||
SaveTheme,
|
||||
)
|
||||
from Users import (
|
||||
AnyUsersExist,
|
||||
CreateUser,
|
||||
GetAvatarPath,
|
||||
ListUsers,
|
||||
UpdateUserColour,
|
||||
UserColourOptions,
|
||||
VerifyLogin,
|
||||
SaveAvatar,
|
||||
)
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
Application = FastAPI(title="SaraView")
|
||||
|
||||
Application.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
allow_credentials=True,
|
||||
)
|
||||
|
||||
|
||||
# --- Auth ---------------------------------------------------------------
|
||||
|
||||
|
||||
class LoginRequest(BaseModel):
|
||||
Username: str
|
||||
Password: str
|
||||
|
||||
|
||||
@Application.get("/api/auth/status")
|
||||
def AuthStatusEndpoint() -> dict:
|
||||
return {"SetupComplete": AnyUsersExist()}
|
||||
|
||||
|
||||
@Application.post("/api/auth/bootstrap")
|
||||
def BootstrapEndpoint(Body: LoginRequest) -> dict:
|
||||
"""Create the first (admin) account. Only usable while no users exist yet,
|
||||
so there's no default/baked-in password sitting in the image.
|
||||
"""
|
||||
if AnyUsersExist():
|
||||
raise HTTPException(status_code=403, detail="Setup already completed.")
|
||||
|
||||
try:
|
||||
CreateUser(Body.Username, Body.Password, True)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Created": True}
|
||||
|
||||
|
||||
@Application.post("/api/auth/login")
|
||||
def LoginEndpoint(Body: LoginRequest, response: Response) -> dict:
|
||||
User = VerifyLogin(Body.Username, Body.Password)
|
||||
|
||||
if User is None:
|
||||
raise HTTPException(status_code=401, detail="Invalid username or password.")
|
||||
|
||||
Token = CreateSession(Body.Username)
|
||||
response.set_cookie(
|
||||
key=SessionCookieName,
|
||||
value=Token,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
max_age=60 * 60 * 24 * 30,
|
||||
)
|
||||
return User
|
||||
|
||||
|
||||
@Application.post("/api/auth/logout")
|
||||
def LogoutEndpoint(request: Request, response: Response) -> dict:
|
||||
Token = request.cookies.get(SessionCookieName)
|
||||
if Token:
|
||||
DeleteSession(Token)
|
||||
response.delete_cookie(SessionCookieName)
|
||||
return {"LoggedOut": True}
|
||||
|
||||
|
||||
@Application.get("/api/auth/me")
|
||||
def MeEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
return CurrentUser
|
||||
|
||||
|
||||
class CreateUserRequest(BaseModel):
|
||||
Username: str
|
||||
Password: str
|
||||
IsAdmin: bool = False
|
||||
Colour: str = "Purple"
|
||||
|
||||
|
||||
@Application.get("/api/users")
|
||||
def GetUsersEndpoint(CurrentUser: dict = Depends(RequireAdmin)) -> list[dict]:
|
||||
return ListUsers()
|
||||
|
||||
|
||||
@Application.post("/api/users")
|
||||
def PostUsersEndpoint(Body: CreateUserRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
|
||||
try:
|
||||
CreateUser(Body.Username, Body.Password, Body.IsAdmin, Body.Colour)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Created": True}
|
||||
|
||||
|
||||
# --- Profile (colour + avatar) --------------------------------------------
|
||||
|
||||
|
||||
@Application.get("/api/profile/colours")
|
||||
def GetProfileColourOptionsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
|
||||
return UserColourOptions
|
||||
|
||||
|
||||
class UpdateColourRequest(BaseModel):
|
||||
Colour: str
|
||||
|
||||
|
||||
@Application.put("/api/profile/colour")
|
||||
def PutProfileColourEndpoint(
|
||||
Body: UpdateColourRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
try:
|
||||
UpdateUserColour(CurrentUser["Username"], Body.Colour)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Username": CurrentUser["Username"], "IsAdmin": CurrentUser["IsAdmin"], "Colour": Body.Colour}
|
||||
|
||||
|
||||
@Application.post("/api/profile/avatar")
|
||||
async def PostProfileAvatarEndpoint(
|
||||
Avatar: UploadFile = File(...), CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
Extension = Path(Avatar.filename or "").suffix.lower() or ".png"
|
||||
FileBytes = await Avatar.read()
|
||||
|
||||
try:
|
||||
SaveAvatar(CurrentUser["Username"], FileBytes, Extension)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Saved": True}
|
||||
|
||||
|
||||
@Application.get("/api/profile/avatar")
|
||||
def GetProfileAvatarEndpoint(Username: str, CurrentUser: dict = Depends(GetCurrentUser)):
|
||||
AvatarPath = GetAvatarPath(Username)
|
||||
|
||||
if AvatarPath is None:
|
||||
raise HTTPException(status_code=404, detail="No avatar set.")
|
||||
|
||||
MimeType = mimetypes.guess_type(str(AvatarPath))[0] or "application/octet-stream"
|
||||
return StreamingResponse(iter([AvatarPath.read_bytes()]), media_type=MimeType)
|
||||
|
||||
|
||||
# --- Files ----------------------------------------------------------------
|
||||
|
||||
|
||||
@Application.get("/api/roots")
|
||||
def GetRootsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
|
||||
return ListRoots()
|
||||
|
||||
|
||||
@Application.get("/api/tree")
|
||||
def GetTreeEndpoint(
|
||||
Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> list[dict]:
|
||||
try:
|
||||
ShowHidden = LoadSettings(CurrentUser["Username"]).get("ShowHiddenFolders", False)
|
||||
return ListDirectory(Root, Path, ShowHidden)
|
||||
except (ValueError, NotADirectoryError, FileNotFoundError) as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
|
||||
class CreateEntryRequest(BaseModel):
|
||||
IsDirectory: bool
|
||||
|
||||
|
||||
@Application.post("/api/tree")
|
||||
def PostTreeEndpoint(
|
||||
Root: str, Path: str, Body: CreateEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
try:
|
||||
CreateEntry(Root, Path, Body.IsDirectory)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
except FileExistsError as Error:
|
||||
raise HTTPException(status_code=409, detail=str(Error))
|
||||
except OSError as Error:
|
||||
raise HTTPException(status_code=500, detail=f"Could not create entry: {Error}")
|
||||
|
||||
return {"Created": True}
|
||||
|
||||
|
||||
class MoveEntryRequest(BaseModel):
|
||||
NewName: str | None = None
|
||||
DestinationPath: str | None = None
|
||||
|
||||
|
||||
@Application.patch("/api/tree")
|
||||
def PatchTreeEndpoint(
|
||||
Root: str, Path: str, Body: MoveEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
if Body.DestinationPath:
|
||||
Destination = Body.DestinationPath
|
||||
elif Body.NewName:
|
||||
ParentPath = Path.rsplit("/", 1)[0] if "/" in Path else ""
|
||||
Destination = f"{ParentPath}/{Body.NewName}" if ParentPath else Body.NewName
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="Provide NewName or DestinationPath.")
|
||||
|
||||
try:
|
||||
MoveEntry(Root, Path, Destination)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
except FileNotFoundError as Error:
|
||||
raise HTTPException(status_code=404, detail=str(Error))
|
||||
except FileExistsError as Error:
|
||||
raise HTTPException(status_code=409, detail=str(Error))
|
||||
except OSError as Error:
|
||||
raise HTTPException(status_code=500, detail=f"Could not move entry: {Error}")
|
||||
|
||||
return {"Moved": True, "NewPath": Destination}
|
||||
|
||||
|
||||
@Application.delete("/api/tree")
|
||||
def DeleteTreeEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
try:
|
||||
DeleteEntry(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
except FileNotFoundError as Error:
|
||||
raise HTTPException(status_code=404, detail=str(Error))
|
||||
except OSError as Error:
|
||||
raise HTTPException(status_code=500, detail=f"Could not delete entry: {Error}")
|
||||
|
||||
return {"Deleted": True}
|
||||
|
||||
|
||||
@Application.get("/api/file")
|
||||
def GetFileEndpoint(
|
||||
Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
):
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
if not FilePath.is_file():
|
||||
raise HTTPException(status_code=404, detail="File not found.")
|
||||
|
||||
FileType = ClassifyFile(FilePath)
|
||||
|
||||
if FileType in TextFileTypes:
|
||||
return PlainTextResponse(ReadTextFile(FilePath))
|
||||
|
||||
FileSize = GetFileSize(FilePath)
|
||||
MimeType = GetMimeType(FilePath)
|
||||
RangeHeader = request.headers.get("range")
|
||||
|
||||
if RangeHeader:
|
||||
Start, End = ParseRangeHeader(RangeHeader, FileSize)
|
||||
ChunkLength = End - Start + 1
|
||||
|
||||
def IterateChunk():
|
||||
with FilePath.open("rb") as SourceFile:
|
||||
SourceFile.seek(Start)
|
||||
yield SourceFile.read(ChunkLength)
|
||||
|
||||
Headers = {
|
||||
"Content-Range": f"bytes {Start}-{End}/{FileSize}",
|
||||
"Accept-Ranges": "bytes",
|
||||
"Content-Length": str(ChunkLength),
|
||||
}
|
||||
return StreamingResponse(
|
||||
IterateChunk(), status_code=206, media_type=MimeType, headers=Headers
|
||||
)
|
||||
|
||||
def IterateFile():
|
||||
with FilePath.open("rb") as SourceFile:
|
||||
while Chunk := SourceFile.read(1024 * 1024):
|
||||
yield Chunk
|
||||
|
||||
Headers = {"Accept-Ranges": "bytes", "Content-Length": str(FileSize)}
|
||||
return StreamingResponse(IterateFile(), media_type=MimeType, headers=Headers)
|
||||
|
||||
|
||||
@Application.put("/api/file")
|
||||
async def PutFileEndpoint(
|
||||
Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
):
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
OldContent = ""
|
||||
if FilePath.is_file():
|
||||
try:
|
||||
OldContent = FilePath.read_text(encoding="utf-8")
|
||||
except (UnicodeDecodeError, OSError):
|
||||
OldContent = ""
|
||||
|
||||
Content = (await request.body()).decode("utf-8")
|
||||
|
||||
try:
|
||||
WriteTextFile(FilePath, Content)
|
||||
except OSError as Error:
|
||||
raise HTTPException(status_code=500, detail=f"Could not write file: {Error}")
|
||||
|
||||
UpdateBlameOnSave(Root, Path, OldContent, Content, CurrentUser["Username"])
|
||||
|
||||
return {"Saved": True}
|
||||
|
||||
|
||||
@Application.get("/api/blame")
|
||||
def GetBlameEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
try:
|
||||
ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return LoadBlame(Root, Path)
|
||||
|
||||
|
||||
@Application.get("/api/raw/{RootName}/{FilePath:path}")
|
||||
def GetRawFileEndpoint(
|
||||
RootName: str, FilePath: str, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
):
|
||||
"""Serves a file at a URL that mirrors its real directory structure
|
||||
(unlike /api/file's query-param form), so relative links, stylesheets,
|
||||
and scripts INSIDE a previewed HTML document resolve correctly against
|
||||
sibling files in the same directory.
|
||||
"""
|
||||
try:
|
||||
ResolvedPath = ResolvePath(RootName, FilePath)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
if not ResolvedPath.is_file():
|
||||
raise HTTPException(status_code=404, detail="File not found.")
|
||||
|
||||
MimeType = mimetypes.guess_type(str(ResolvedPath))[0] or "application/octet-stream"
|
||||
|
||||
def IterateFile():
|
||||
with ResolvedPath.open("rb") as SourceFile:
|
||||
while Chunk := SourceFile.read(1024 * 1024):
|
||||
yield Chunk
|
||||
|
||||
return StreamingResponse(IterateFile(), media_type=MimeType)
|
||||
|
||||
|
||||
@Application.get("/api/book")
|
||||
def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
Book = ReadBookForFile(FilePath)
|
||||
return Book or {}
|
||||
|
||||
|
||||
# --- Settings / themes ------------------------------------------------------
|
||||
|
||||
|
||||
@Application.get("/api/settings")
|
||||
def GetSettingsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
return LoadSettings(CurrentUser["Username"])
|
||||
|
||||
|
||||
@Application.put("/api/settings")
|
||||
def PutSettingsEndpoint(NewSettings: dict, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
return SaveSettings(CurrentUser["Username"], NewSettings)
|
||||
|
||||
|
||||
@Application.get("/api/themes")
|
||||
def GetThemesEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
|
||||
return ListThemeNames()
|
||||
|
||||
|
||||
@Application.get("/api/theme")
|
||||
def GetThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
try:
|
||||
return LoadTheme(Name)
|
||||
except FileNotFoundError as Error:
|
||||
raise HTTPException(status_code=404, detail=str(Error))
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
|
||||
class SaveThemeRequest(BaseModel):
|
||||
Palette: dict[str, str]
|
||||
|
||||
|
||||
@Application.put("/api/theme")
|
||||
def PutThemeEndpoint(
|
||||
Name: str, Body: SaveThemeRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
try:
|
||||
SaveTheme(Name, Body.Palette)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Saved": True, "Roles": PaletteRoles}
|
||||
|
||||
|
||||
@Application.delete("/api/theme")
|
||||
def DeleteThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
try:
|
||||
DeleteTheme(Name)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
except FileNotFoundError as Error:
|
||||
raise HTTPException(status_code=404, detail=str(Error))
|
||||
|
||||
return {"Deleted": True}
|
||||
|
||||
|
||||
# In production, the frontend's built assets are copied alongside this file
|
||||
# under Frontend/dist (see Dockerfile) and served directly by the backend.
|
||||
# This stays unauthenticated — the SPA itself needs to load before it can
|
||||
# even show a login screen.
|
||||
FrontendDist = ApplicationDirectory.parent / "Frontend" / "dist"
|
||||
if FrontendDist.is_dir():
|
||||
Application.mount("/", StaticFiles(directory=str(FrontendDist), html=True), name="Frontend")
|
||||
@@ -0,0 +1,44 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
TextFileTypes = {"markdown", "text", "config", "html"}
|
||||
|
||||
MimeTypeMap = {
|
||||
".png": "image/png",
|
||||
".jpg": "image/jpeg",
|
||||
".jpeg": "image/jpeg",
|
||||
".gif": "image/gif",
|
||||
".webp": "image/webp",
|
||||
".svg": "image/svg+xml",
|
||||
".mp4": "video/mp4",
|
||||
".webm": "video/webm",
|
||||
".mov": "video/quicktime",
|
||||
".mkv": "video/x-matroska",
|
||||
}
|
||||
|
||||
|
||||
def ReadTextFile(FilePath: Path) -> str:
|
||||
"""Read a text-classified file as UTF-8."""
|
||||
return FilePath.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def GetFileSize(FilePath: Path) -> int:
|
||||
return FilePath.stat().st_size
|
||||
|
||||
|
||||
def GetMimeType(FilePath: Path) -> str:
|
||||
return MimeTypeMap.get(FilePath.suffix.lower(), "application/octet-stream")
|
||||
|
||||
|
||||
def ParseRangeHeader(RangeHeader: str, FileSize: int) -> tuple[int, int]:
|
||||
"""Parse a single-range HTTP Range header into an inclusive (Start, End)."""
|
||||
RangeValue = RangeHeader.replace("bytes=", "").strip()
|
||||
StartText, _, EndText = RangeValue.partition("-")
|
||||
|
||||
Start = int(StartText) if StartText else 0
|
||||
End = int(EndText) if EndText else FileSize - 1
|
||||
End = min(End, FileSize - 1)
|
||||
|
||||
return Start, End
|
||||
@@ -0,0 +1,33 @@
|
||||
# SaraView configuration.
|
||||
# Paths here are resolved as given — use absolute container paths when running
|
||||
# via Docker (see docker-compose.yml), matching whatever you mount the volumes to.
|
||||
|
||||
[SaraView]
|
||||
ConfigVersion = 1
|
||||
ApplicationVersion = "0.1.0-beta"
|
||||
|
||||
[Server]
|
||||
Host = "0.0.0.0"
|
||||
Port = 8420
|
||||
|
||||
# Where SaraView keeps its own data — Settings.toml and the Themes/ folder.
|
||||
# This is NOT one of the [Roots] below, so it is structurally unreachable
|
||||
# from the file browser regardless of any hidden-file setting. Give it its
|
||||
# own persistent volume (see docker-compose.yml) so it survives rebuilds.
|
||||
[Storage]
|
||||
DataDirectory = "/data/SaraViewData"
|
||||
|
||||
# Named roots the browser can navigate. Add as many as you like; each becomes
|
||||
# a top-level entry point in the frontend's root switcher.
|
||||
[Roots]
|
||||
Notes = "/data/notes"
|
||||
|
||||
# Integration with SaraTag: if Enabled, SaraView imports SaraTag's own Core.py
|
||||
# directly (by file path, not by installing it as a package) to resolve Book
|
||||
# tag files exactly the way SaraTag itself does. This is read-only — SaraView
|
||||
# never writes Book or Lexicon files.
|
||||
[SaraTag]
|
||||
Enabled = true
|
||||
CoreModulePath = "/data/SaraTag/.saratag"
|
||||
HideInternalFiles = true
|
||||
InternalPatterns = [".saratag", ".SaraTag*", "*.book.toml"]
|
||||
@@ -0,0 +1,185 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import tomllib
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
|
||||
from Core import ApplicationDirectory, Config
|
||||
|
||||
|
||||
PaletteRoles = [
|
||||
"Background",
|
||||
"Surface",
|
||||
"SurfaceAlt",
|
||||
"Border",
|
||||
"Text",
|
||||
"TextMuted",
|
||||
"Black",
|
||||
"White",
|
||||
"Red",
|
||||
"Orange",
|
||||
"Yellow",
|
||||
"Green",
|
||||
"Blue",
|
||||
"Purple",
|
||||
]
|
||||
|
||||
DefaultSettings: dict[str, Any] = {
|
||||
"FontFamily": "Fixedsys Excelsior",
|
||||
"FontSize": 14,
|
||||
"ThemeName": "CatppuccinMocha",
|
||||
"ShowHiddenFolders": False,
|
||||
"ShowBreadcrumbs": True,
|
||||
"UseHyphenBullets": True,
|
||||
"UseAsciiFolderIcons": True,
|
||||
"NavigateIntoDirectories": False,
|
||||
"ShowGitBlame": False,
|
||||
"NewEntryButtonsAtTop": False,
|
||||
"StickyToolbar": True,
|
||||
"ConfirmBeforeDelete": True,
|
||||
}
|
||||
|
||||
DefaultThemesDirectory = ApplicationDirectory / "DefaultThemes"
|
||||
|
||||
|
||||
def SanitizeThemeName(ThemeName: str) -> str:
|
||||
"""Reject theme names that could escape the Themes directory."""
|
||||
if not ThemeName or "/" in ThemeName or "\\" in ThemeName or ThemeName in (".", ".."):
|
||||
raise ValueError(f"Invalid theme name: {ThemeName}")
|
||||
|
||||
return ThemeName
|
||||
|
||||
|
||||
def SanitizeUsernameForPath(Username: str) -> str:
|
||||
"""Usernames are validated at account-creation time, but this is a second
|
||||
line of defense against a settings path ever escaping its directory.
|
||||
"""
|
||||
if not Username or not re.fullmatch(r"[A-Za-z0-9_.-]+", Username):
|
||||
raise ValueError(f"Invalid username: {Username}")
|
||||
|
||||
return Username
|
||||
|
||||
|
||||
def GetDataDirectory() -> Path:
|
||||
StorageConfig = Config.get("Storage", {})
|
||||
RawPath = StorageConfig.get("DataDirectory", str(ApplicationDirectory / "Data"))
|
||||
return Path(RawPath).expanduser().resolve()
|
||||
|
||||
|
||||
def GetSettingsDirectory() -> Path:
|
||||
return GetDataDirectory() / "Settings"
|
||||
|
||||
|
||||
def GetSettingsPath(Username: str) -> Path:
|
||||
Username = SanitizeUsernameForPath(Username)
|
||||
return GetSettingsDirectory() / f"{Username}.toml"
|
||||
|
||||
|
||||
def GetThemesDirectory() -> Path:
|
||||
return GetDataDirectory() / "Themes"
|
||||
|
||||
|
||||
def EnsureDataDirectory() -> None:
|
||||
"""Create the data directory and seed default themes, if missing."""
|
||||
DataDirectory = GetDataDirectory()
|
||||
ThemesDirectory = GetThemesDirectory()
|
||||
DataDirectory.mkdir(parents=True, exist_ok=True)
|
||||
GetSettingsDirectory().mkdir(parents=True, exist_ok=True)
|
||||
ThemesDirectory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if DefaultThemesDirectory.is_dir():
|
||||
for ThemeFile in DefaultThemesDirectory.glob("*.toml"):
|
||||
TargetPath = ThemesDirectory / ThemeFile.name
|
||||
if not TargetPath.is_file():
|
||||
shutil.copy(ThemeFile, TargetPath)
|
||||
|
||||
|
||||
def LoadSettings(Username: str) -> dict[str, Any]:
|
||||
"""Load one user's settings, seeding their file with the defaults on
|
||||
first use. Settings are per-user; there is no shared/global settings
|
||||
file anymore.
|
||||
"""
|
||||
EnsureDataDirectory()
|
||||
SettingsPath = GetSettingsPath(Username)
|
||||
|
||||
if not SettingsPath.is_file():
|
||||
with SettingsPath.open("wb") as SettingsFile:
|
||||
tomli_w.dump(DefaultSettings, SettingsFile)
|
||||
return dict(DefaultSettings)
|
||||
|
||||
with SettingsPath.open("rb") as SettingsFile:
|
||||
Loaded = tomllib.load(SettingsFile)
|
||||
|
||||
return {**DefaultSettings, **Loaded}
|
||||
|
||||
|
||||
def SaveSettings(Username: str, NewSettings: dict[str, Any]) -> dict[str, Any]:
|
||||
EnsureDataDirectory()
|
||||
SettingsPath = GetSettingsPath(Username)
|
||||
Current: dict[str, Any] = {}
|
||||
|
||||
if SettingsPath.is_file():
|
||||
with SettingsPath.open("rb") as SettingsFile:
|
||||
Current = tomllib.load(SettingsFile)
|
||||
|
||||
Merged = {**DefaultSettings, **Current, **NewSettings}
|
||||
|
||||
with SettingsPath.open("wb") as SettingsFile:
|
||||
tomli_w.dump(Merged, SettingsFile)
|
||||
|
||||
return Merged
|
||||
|
||||
|
||||
def ListThemeNames() -> list[str]:
|
||||
EnsureDataDirectory()
|
||||
return sorted(ThemeFile.stem for ThemeFile in GetThemesDirectory().glob("*.toml"))
|
||||
|
||||
|
||||
def DeleteTheme(ThemeName: str) -> None:
|
||||
EnsureDataDirectory()
|
||||
ThemeName = SanitizeThemeName(ThemeName)
|
||||
ThemePath = GetThemesDirectory() / f"{ThemeName}.toml"
|
||||
|
||||
if not ThemePath.is_file():
|
||||
raise FileNotFoundError(f"Unknown theme: {ThemeName}")
|
||||
|
||||
ThemePath.unlink()
|
||||
|
||||
|
||||
def LoadTheme(ThemeName: str) -> dict[str, str]:
|
||||
EnsureDataDirectory()
|
||||
ThemeName = SanitizeThemeName(ThemeName)
|
||||
ThemePath = GetThemesDirectory() / f"{ThemeName}.toml"
|
||||
|
||||
if not ThemePath.is_file():
|
||||
raise FileNotFoundError(f"Unknown theme: {ThemeName}")
|
||||
|
||||
with ThemePath.open("rb") as ThemeFile:
|
||||
Loaded = tomllib.load(ThemeFile)
|
||||
|
||||
return Loaded.get("Palette", {})
|
||||
|
||||
|
||||
def SaveTheme(ThemeName: str, Palette: dict[str, str]) -> None:
|
||||
"""Create or overwrite a custom theme. Missing roles fall back to the
|
||||
active Catppuccin Mocha default so a partially-specified custom theme
|
||||
still renders sensibly.
|
||||
"""
|
||||
EnsureDataDirectory()
|
||||
ThemeName = SanitizeThemeName(ThemeName)
|
||||
|
||||
MissingRoles = [Role for Role in PaletteRoles if Role not in Palette]
|
||||
if MissingRoles:
|
||||
FallbackPalette = LoadTheme("CatppuccinMocha")
|
||||
for Role in MissingRoles:
|
||||
Palette[Role] = FallbackPalette.get(Role, "#ff00ff")
|
||||
|
||||
ThemePath = GetThemesDirectory() / f"{ThemeName}.toml"
|
||||
|
||||
with ThemePath.open("wb") as ThemeFile:
|
||||
tomli_w.dump({"Palette": Palette}, ThemeFile)
|
||||
@@ -0,0 +1,170 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import secrets
|
||||
import tomllib
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
|
||||
from Settings import GetDataDirectory
|
||||
|
||||
|
||||
Pbkdf2Iterations = 260_000
|
||||
|
||||
UserColourOptions = ["Red", "Orange", "Yellow", "Green", "Blue", "Purple"]
|
||||
DefaultUserColour = "Purple"
|
||||
|
||||
AllowedAvatarExtensions = {".png", ".jpg", ".jpeg", ".gif", ".webp"}
|
||||
|
||||
|
||||
def GetUsersPath():
|
||||
return GetDataDirectory() / "Users.toml"
|
||||
|
||||
|
||||
def GetAvatarDirectory() -> Path:
|
||||
return GetDataDirectory() / "Avatars"
|
||||
|
||||
|
||||
def HashPassword(Password: str) -> str:
|
||||
Salt = secrets.token_hex(16)
|
||||
DerivedKey = hashlib.pbkdf2_hmac(
|
||||
"sha256", Password.encode("utf-8"), bytes.fromhex(Salt), Pbkdf2Iterations
|
||||
)
|
||||
return f"pbkdf2_sha256${Pbkdf2Iterations}${Salt}${DerivedKey.hex()}"
|
||||
|
||||
|
||||
def VerifyPassword(Password: str, StoredHash: str) -> bool:
|
||||
try:
|
||||
Algorithm, IterationsText, Salt, HashHex = StoredHash.split("$")
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
if Algorithm != "pbkdf2_sha256":
|
||||
return False
|
||||
|
||||
DerivedKey = hashlib.pbkdf2_hmac(
|
||||
"sha256", Password.encode("utf-8"), bytes.fromhex(Salt), int(IterationsText)
|
||||
)
|
||||
return hmac.compare_digest(DerivedKey.hex(), HashHex)
|
||||
|
||||
|
||||
def LoadUsers() -> dict[str, dict[str, Any]]:
|
||||
UsersPath = GetUsersPath()
|
||||
|
||||
if not UsersPath.is_file():
|
||||
return {}
|
||||
|
||||
with UsersPath.open("rb") as UsersFile:
|
||||
Loaded = tomllib.load(UsersFile)
|
||||
|
||||
return Loaded.get("Users", {})
|
||||
|
||||
|
||||
def SaveUsers(Users: dict[str, dict[str, Any]]) -> None:
|
||||
GetDataDirectory().mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with GetUsersPath().open("wb") as UsersFile:
|
||||
tomli_w.dump({"Users": Users}, UsersFile)
|
||||
|
||||
|
||||
def AnyUsersExist() -> bool:
|
||||
return len(LoadUsers()) > 0
|
||||
|
||||
|
||||
def GetUser(Username: str) -> dict[str, Any] | None:
|
||||
return LoadUsers().get(Username)
|
||||
|
||||
|
||||
def CreateUser(Username: str, Password: str, IsAdmin: bool, Colour: str = DefaultUserColour) -> None:
|
||||
if not Username or not Username.strip():
|
||||
raise ValueError("Username cannot be empty.")
|
||||
|
||||
if not Password:
|
||||
raise ValueError("Password cannot be empty.")
|
||||
|
||||
if Colour not in UserColourOptions:
|
||||
Colour = DefaultUserColour
|
||||
|
||||
Users = LoadUsers()
|
||||
|
||||
if Username in Users:
|
||||
raise ValueError(f"User '{Username}' already exists.")
|
||||
|
||||
Users[Username] = {
|
||||
"PasswordHash": HashPassword(Password),
|
||||
"IsAdmin": IsAdmin,
|
||||
"Colour": Colour,
|
||||
}
|
||||
SaveUsers(Users)
|
||||
|
||||
|
||||
def UpdateUserColour(Username: str, Colour: str) -> None:
|
||||
if Colour not in UserColourOptions:
|
||||
raise ValueError(f"Invalid colour: {Colour}")
|
||||
|
||||
Users = LoadUsers()
|
||||
|
||||
if Username not in Users:
|
||||
raise ValueError("User not found.")
|
||||
|
||||
Users[Username]["Colour"] = Colour
|
||||
SaveUsers(Users)
|
||||
|
||||
|
||||
def VerifyLogin(Username: str, Password: str) -> dict[str, Any] | None:
|
||||
User = GetUser(Username)
|
||||
|
||||
if User is None:
|
||||
return None
|
||||
|
||||
if not VerifyPassword(Password, User["PasswordHash"]):
|
||||
return None
|
||||
|
||||
return {
|
||||
"Username": Username,
|
||||
"IsAdmin": User.get("IsAdmin", False),
|
||||
"Colour": User.get("Colour", DefaultUserColour),
|
||||
}
|
||||
|
||||
|
||||
def ListUsers() -> list[dict[str, Any]]:
|
||||
Users = LoadUsers()
|
||||
return [
|
||||
{
|
||||
"Username": Name,
|
||||
"IsAdmin": Data.get("IsAdmin", False),
|
||||
"Colour": Data.get("Colour", DefaultUserColour),
|
||||
}
|
||||
for Name, Data in Users.items()
|
||||
]
|
||||
|
||||
|
||||
def SaveAvatar(Username: str, FileBytes: bytes, Extension: str) -> None:
|
||||
Extension = Extension.lower()
|
||||
|
||||
if Extension not in AllowedAvatarExtensions:
|
||||
raise ValueError(f"Unsupported image type: {Extension}")
|
||||
|
||||
Directory = GetAvatarDirectory()
|
||||
Directory.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Remove any previous avatar for this user under a different extension.
|
||||
for ExistingPath in Directory.glob(f"{Username}.*"):
|
||||
ExistingPath.unlink()
|
||||
|
||||
AvatarPath = Directory / f"{Username}{Extension}"
|
||||
AvatarPath.write_bytes(FileBytes)
|
||||
|
||||
|
||||
def GetAvatarPath(Username: str) -> Path | None:
|
||||
Directory = GetAvatarDirectory()
|
||||
|
||||
if not Directory.is_dir():
|
||||
return None
|
||||
|
||||
Matches = sorted(Directory.glob(f"{Username}.*"))
|
||||
return Matches[0] if Matches else None
|
||||
@@ -0,0 +1,4 @@
|
||||
fastapi==0.115.0
|
||||
uvicorn[standard]==0.30.6
|
||||
tomli-w==1.0.0
|
||||
python-multipart==0.0.12
|
||||
Reference in new issue
Block a user