mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-14 02:38:11 +02:00
Added groups and restricted access.
This commit is contained in:
1 parent
8aa223bc7b
commit
9a5b9530a3
17 files changed
+1406
-66
No files matched your search
+7
-1
@@ -6,6 +6,7 @@ from typing import Any
|
||||
|
||||
from fastapi import Depends, HTTPException, Request
|
||||
|
||||
from Groups import AdminGroupName, GetGroupsForUser, IsUserInGroup
|
||||
from Users import GetUser
|
||||
|
||||
|
||||
@@ -39,9 +40,14 @@ def GetCurrentUser(request: Request) -> dict[str, Any]:
|
||||
if User is None:
|
||||
raise HTTPException(status_code=401, detail="Session user no longer exists.")
|
||||
|
||||
# Admin isn't a stored flag — it's membership in the built-in Admin
|
||||
# group, computed live, so it behaves exactly like any other group.
|
||||
UserGroups = GetGroupsForUser(Username)
|
||||
|
||||
return {
|
||||
"Username": Username,
|
||||
"IsAdmin": User.get("IsAdmin", False),
|
||||
"IsAdmin": AdminGroupName in UserGroups,
|
||||
"Groups": UserGroups,
|
||||
"Colour": User.get("Colour", "Purple"),
|
||||
}
|
||||
|
||||
|
||||
@@ -44,6 +44,33 @@ def SaveBlame(RootName: str, RelativePath: str, Blame: dict[str, dict[str, Any]]
|
||||
tomli_w.dump({"Lines": Blame}, BlameFile)
|
||||
|
||||
|
||||
def RenameUserInBlame(OldUsername: str, NewUsername: str) -> None:
|
||||
"""Rewrite every recorded blame attribution from OldUsername to
|
||||
NewUsername across every file's blame history — called when a user
|
||||
renames their account, so past attributions stay coherent.
|
||||
"""
|
||||
BlameDirectory = GetBlameDirectory()
|
||||
|
||||
if not BlameDirectory.is_dir():
|
||||
return
|
||||
|
||||
for BlamePath in BlameDirectory.rglob("*.blame.toml"):
|
||||
with BlamePath.open("rb") as BlameFile:
|
||||
Loaded = tomllib.load(BlameFile)
|
||||
|
||||
Lines = Loaded.get("Lines", {})
|
||||
Changed = False
|
||||
|
||||
for Entry in Lines.values():
|
||||
if Entry.get("Username") == OldUsername:
|
||||
Entry["Username"] = NewUsername
|
||||
Changed = True
|
||||
|
||||
if Changed:
|
||||
with BlamePath.open("wb") as BlameFile:
|
||||
tomli_w.dump({"Lines": Lines}, BlameFile)
|
||||
|
||||
|
||||
def UpdateBlameOnSave(
|
||||
RootName: str, RelativePath: str, OldContent: str, NewContent: str, Username: str
|
||||
) -> None:
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tomllib
|
||||
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
|
||||
from Settings import GetDataDirectory
|
||||
|
||||
|
||||
AdminGroupName = "Admin"
|
||||
|
||||
|
||||
def GetGroupsPath():
|
||||
return GetDataDirectory() / "Groups.toml"
|
||||
|
||||
|
||||
def LoadGroups() -> dict[str, list[str]]:
|
||||
GroupsPath = GetGroupsPath()
|
||||
|
||||
if not GroupsPath.is_file():
|
||||
# First run after upgrading from the old boolean-flag admin model:
|
||||
# seed the Admin group from whoever already had IsAdmin set, so
|
||||
# nobody silently loses admin access.
|
||||
from Users import LoadUsers
|
||||
|
||||
SeededAdmins = [
|
||||
Username for Username, Data in LoadUsers().items() if Data.get("IsAdmin", False)
|
||||
]
|
||||
Groups = {AdminGroupName: SeededAdmins}
|
||||
SaveGroups(Groups)
|
||||
return Groups
|
||||
|
||||
with GroupsPath.open("rb") as GroupsFile:
|
||||
Loaded = tomllib.load(GroupsFile)
|
||||
|
||||
Groups = Loaded.get("Groups", {})
|
||||
|
||||
if AdminGroupName not in Groups:
|
||||
Groups[AdminGroupName] = []
|
||||
SaveGroups(Groups)
|
||||
|
||||
return Groups
|
||||
|
||||
|
||||
def SaveGroups(Groups: dict[str, list[str]]) -> None:
|
||||
GetDataDirectory().mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with GetGroupsPath().open("wb") as GroupsFile:
|
||||
tomli_w.dump({"Groups": Groups}, GroupsFile)
|
||||
|
||||
|
||||
def ListGroups() -> list[dict[str, Any]]:
|
||||
Groups = LoadGroups()
|
||||
return [{"Name": Name, "Members": Members} for Name, Members in sorted(Groups.items())]
|
||||
|
||||
|
||||
def CreateGroup(Name: str) -> None:
|
||||
Name = Name.strip()
|
||||
|
||||
if not Name:
|
||||
raise ValueError("Group name cannot be empty.")
|
||||
|
||||
Groups = LoadGroups()
|
||||
|
||||
if Name in Groups:
|
||||
raise ValueError(f"Group '{Name}' already exists.")
|
||||
|
||||
Groups[Name] = []
|
||||
SaveGroups(Groups)
|
||||
|
||||
|
||||
def DeleteGroup(Name: str) -> None:
|
||||
if Name == AdminGroupName:
|
||||
raise ValueError("The Admin group can't be deleted.")
|
||||
|
||||
Groups = LoadGroups()
|
||||
|
||||
if Name not in Groups:
|
||||
raise ValueError(f"Group '{Name}' does not exist.")
|
||||
|
||||
del Groups[Name]
|
||||
SaveGroups(Groups)
|
||||
|
||||
|
||||
def SetGroupMembers(Name: str, Members: list[str]) -> None:
|
||||
Groups = LoadGroups()
|
||||
|
||||
if Name not in Groups:
|
||||
raise ValueError(f"Group '{Name}' does not exist.")
|
||||
|
||||
# De-duplicate while keeping things predictable.
|
||||
Groups[Name] = sorted(set(Members))
|
||||
SaveGroups(Groups)
|
||||
|
||||
|
||||
def AddMember(GroupName: str, Username: str) -> None:
|
||||
Groups = LoadGroups()
|
||||
|
||||
if GroupName not in Groups:
|
||||
raise ValueError(f"Group '{GroupName}' does not exist.")
|
||||
|
||||
if Username not in Groups[GroupName]:
|
||||
Groups[GroupName].append(Username)
|
||||
SaveGroups(Groups)
|
||||
|
||||
|
||||
def RemoveMember(GroupName: str, Username: str) -> None:
|
||||
Groups = LoadGroups()
|
||||
|
||||
if GroupName not in Groups:
|
||||
raise ValueError(f"Group '{GroupName}' does not exist.")
|
||||
|
||||
if Username in Groups[GroupName]:
|
||||
Groups[GroupName].remove(Username)
|
||||
SaveGroups(Groups)
|
||||
|
||||
|
||||
def GetGroupsForUser(Username: str) -> list[str]:
|
||||
Groups = LoadGroups()
|
||||
return [Name for Name, Members in Groups.items() if Username in Members]
|
||||
|
||||
|
||||
def IsUserInGroup(Username: str, GroupName: str) -> bool:
|
||||
return Username in LoadGroups().get(GroupName, [])
|
||||
|
||||
|
||||
def RenameUserInGroups(OldUsername: str, NewUsername: str) -> None:
|
||||
Groups = LoadGroups()
|
||||
Changed = False
|
||||
|
||||
for Members in Groups.values():
|
||||
if OldUsername in Members:
|
||||
Members.remove(OldUsername)
|
||||
Members.append(NewUsername)
|
||||
Changed = True
|
||||
|
||||
if Changed:
|
||||
SaveGroups(Groups)
|
||||
@@ -0,0 +1,124 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import tomllib
|
||||
|
||||
from typing import Any
|
||||
|
||||
import tomli_w
|
||||
from fastapi import HTTPException
|
||||
|
||||
from Settings import GetDataDirectory
|
||||
|
||||
|
||||
def GetLocksPath():
|
||||
return GetDataDirectory() / "LockedDirectories.toml"
|
||||
|
||||
|
||||
def LoadLocks() -> list[dict[str, Any]]:
|
||||
LocksPath = GetLocksPath()
|
||||
|
||||
if not LocksPath.is_file():
|
||||
return []
|
||||
|
||||
with LocksPath.open("rb") as LocksFile:
|
||||
Loaded = tomllib.load(LocksFile)
|
||||
|
||||
return Loaded.get("Locks", [])
|
||||
|
||||
|
||||
def SaveLocks(Locks: list[dict[str, Any]]) -> None:
|
||||
GetDataDirectory().mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with GetLocksPath().open("wb") as LocksFile:
|
||||
tomli_w.dump({"Locks": Locks}, LocksFile)
|
||||
|
||||
|
||||
def GetLockForPath(RootName: str, RelativePath: str) -> dict[str, Any] | None:
|
||||
for Lock in LoadLocks():
|
||||
if Lock.get("Root") == RootName and Lock.get("Path") == RelativePath:
|
||||
return Lock
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def SetLock(RootName: str, RelativePath: str, Usernames: list[str], GroupNames: list[str]) -> None:
|
||||
"""Restrict a file or directory to specific users and/or groups. Passing
|
||||
both empty lists is the same as removing the lock entirely.
|
||||
"""
|
||||
if not RelativePath:
|
||||
raise ValueError("Can't lock an entire root — pick a specific file or directory.")
|
||||
|
||||
if not Usernames and not GroupNames:
|
||||
RemoveLock(RootName, RelativePath)
|
||||
return
|
||||
|
||||
Locks = LoadLocks()
|
||||
|
||||
for Lock in Locks:
|
||||
if Lock.get("Root") == RootName and Lock.get("Path") == RelativePath:
|
||||
Lock["Users"] = Usernames
|
||||
Lock["Groups"] = GroupNames
|
||||
SaveLocks(Locks)
|
||||
return
|
||||
|
||||
Locks.append({"Root": RootName, "Path": RelativePath, "Users": Usernames, "Groups": GroupNames})
|
||||
SaveLocks(Locks)
|
||||
|
||||
|
||||
def RemoveLock(RootName: str, RelativePath: str) -> None:
|
||||
Locks = [
|
||||
Lock
|
||||
for Lock in LoadLocks()
|
||||
if not (Lock.get("Root") == RootName and Lock.get("Path") == RelativePath)
|
||||
]
|
||||
SaveLocks(Locks)
|
||||
|
||||
|
||||
def IsPathAccessible(
|
||||
RootName: str, RelativePath: str, Username: str, UserGroups: list[str], IsAdmin: bool
|
||||
) -> bool:
|
||||
"""A path is inaccessible if it IS a locked file/directory, or lies
|
||||
anywhere underneath a locked directory, and the user isn't permitted
|
||||
(by username or group) on every applicable lock. Locks don't propagate
|
||||
to subdirectories as separate records — there's only ever one lock
|
||||
entry per locked directory — but access to anything inside still
|
||||
requires being permitted on that ancestor lock, so you always have to
|
||||
"enter the parent first." Admins bypass every lock.
|
||||
"""
|
||||
if IsAdmin:
|
||||
return True
|
||||
|
||||
for Lock in LoadLocks():
|
||||
if Lock.get("Root") != RootName:
|
||||
continue
|
||||
|
||||
LockPath = Lock.get("Path", "")
|
||||
|
||||
if not LockPath:
|
||||
continue
|
||||
|
||||
if RelativePath == LockPath or RelativePath.startswith(f"{LockPath}/"):
|
||||
PermittedUsers = Lock.get("Users", [])
|
||||
PermittedGroups = Lock.get("Groups", [])
|
||||
IsPermittedByUser = Username in PermittedUsers
|
||||
IsPermittedByGroup = any(Group in PermittedGroups for Group in UserGroups)
|
||||
|
||||
if not IsPermittedByUser and not IsPermittedByGroup:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def RequirePathAccessible(RootName: str, RelativePath: str, CurrentUser: dict[str, Any]) -> None:
|
||||
"""Raises a 404 (never a distinguishing "access denied") for a path the
|
||||
current user isn't permitted to see, so a locked directory's existence
|
||||
isn't revealed to anyone not on its list.
|
||||
"""
|
||||
if not IsPathAccessible(
|
||||
RootName,
|
||||
RelativePath,
|
||||
CurrentUser["Username"],
|
||||
CurrentUser.get("Groups", []),
|
||||
CurrentUser.get("IsAdmin", False),
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="Not found.")
|
||||
+247
-6
@@ -15,10 +15,25 @@ from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from Auth import CreateSession, DeleteSession, GetCurrentUser, RequireAdmin, SessionCookieName
|
||||
from BookReader import ReadBookForFile
|
||||
from Blame import LoadBlame, UpdateBlameOnSave
|
||||
from Blame import LoadBlame, RenameUserInBlame, UpdateBlameOnSave
|
||||
from Core import ResolvePath
|
||||
from Editor import WriteTextFile
|
||||
from Explorer import ClassifyFile, CreateEntry, DeleteEntry, ListDirectory, ListRoots, MoveEntry
|
||||
from Groups import (
|
||||
CreateGroup,
|
||||
DeleteGroup,
|
||||
ListGroups,
|
||||
RenameUserInGroups,
|
||||
SetGroupMembers,
|
||||
)
|
||||
from Locks import (
|
||||
GetLockForPath,
|
||||
IsPathAccessible,
|
||||
LoadLocks,
|
||||
RemoveLock,
|
||||
RequirePathAccessible,
|
||||
SetLock,
|
||||
)
|
||||
from Preview import GetFileSize, GetMimeType, ParseRangeHeader, ReadTextFile, TextFileTypes
|
||||
from Settings import (
|
||||
DeleteTheme,
|
||||
@@ -26,18 +41,22 @@ from Settings import (
|
||||
LoadSettings,
|
||||
LoadTheme,
|
||||
PaletteRoles,
|
||||
RenameUserSettings,
|
||||
SaveSettings,
|
||||
SaveTheme,
|
||||
)
|
||||
from Users import (
|
||||
AnyUsersExist,
|
||||
ChangePassword,
|
||||
CreateUser,
|
||||
GetAvatarPath,
|
||||
ListUsers,
|
||||
RenameUser,
|
||||
RenameUserAvatar,
|
||||
SaveAvatar,
|
||||
UpdateUserColour,
|
||||
UserColourOptions,
|
||||
VerifyLogin,
|
||||
SaveAvatar,
|
||||
)
|
||||
from pydantic import BaseModel
|
||||
|
||||
@@ -122,7 +141,7 @@ class CreateUserRequest(BaseModel):
|
||||
|
||||
|
||||
@Application.get("/api/users")
|
||||
def GetUsersEndpoint(CurrentUser: dict = Depends(RequireAdmin)) -> list[dict]:
|
||||
def GetUsersEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
|
||||
return ListUsers()
|
||||
|
||||
|
||||
@@ -136,7 +155,7 @@ def PostUsersEndpoint(Body: CreateUserRequest, CurrentUser: dict = Depends(Requi
|
||||
return {"Created": True}
|
||||
|
||||
|
||||
# --- Profile (colour + avatar) --------------------------------------------
|
||||
# --- Profile (colour, avatar, username, password) --------------------------
|
||||
|
||||
|
||||
@Application.get("/api/profile/colours")
|
||||
@@ -186,6 +205,163 @@ def GetProfileAvatarEndpoint(Username: str, CurrentUser: dict = Depends(GetCurre
|
||||
return StreamingResponse(iter([AvatarPath.read_bytes()]), media_type=MimeType)
|
||||
|
||||
|
||||
class ChangeUsernameRequest(BaseModel):
|
||||
NewUsername: str
|
||||
|
||||
|
||||
@Application.put("/api/profile/username")
|
||||
def PutProfileUsernameEndpoint(
|
||||
Body: ChangeUsernameRequest,
|
||||
request: Request,
|
||||
response: Response,
|
||||
CurrentUser: dict = Depends(GetCurrentUser),
|
||||
) -> dict:
|
||||
OldUsername = CurrentUser["Username"]
|
||||
NewUsername = Body.NewUsername.strip()
|
||||
|
||||
try:
|
||||
RenameUser(OldUsername, NewUsername)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
if NewUsername != OldUsername:
|
||||
RenameUserSettings(OldUsername, NewUsername)
|
||||
RenameUserAvatar(OldUsername, NewUsername)
|
||||
RenameUserInBlame(OldUsername, NewUsername)
|
||||
RenameUserInGroups(OldUsername, NewUsername)
|
||||
|
||||
# Re-authenticate this same request under the new name so the
|
||||
# person doesn't get logged out by their own rename. Any OTHER
|
||||
# active session for the old name is left to fail naturally on its
|
||||
# next request — a re-login there is the simplest correct outcome.
|
||||
OldToken = request.cookies.get(SessionCookieName)
|
||||
if OldToken:
|
||||
DeleteSession(OldToken)
|
||||
|
||||
NewToken = CreateSession(NewUsername)
|
||||
response.set_cookie(
|
||||
key=SessionCookieName,
|
||||
value=NewToken,
|
||||
httponly=True,
|
||||
samesite="lax",
|
||||
max_age=60 * 60 * 24 * 30,
|
||||
)
|
||||
|
||||
return {"Username": NewUsername, "IsAdmin": CurrentUser["IsAdmin"], "Colour": CurrentUser["Colour"]}
|
||||
|
||||
|
||||
class ChangePasswordRequest(BaseModel):
|
||||
CurrentPassword: str
|
||||
NewPassword: str
|
||||
|
||||
|
||||
@Application.put("/api/profile/password")
|
||||
def PutProfilePasswordEndpoint(
|
||||
Body: ChangePasswordRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
try:
|
||||
ChangePassword(CurrentUser["Username"], Body.CurrentPassword, Body.NewPassword)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Changed": True}
|
||||
|
||||
|
||||
# --- Groups -----------------------------------------------------------------
|
||||
|
||||
|
||||
@Application.get("/api/groups")
|
||||
def GetGroupsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
|
||||
# Readable by anyone — picking an existing group as a lock target is
|
||||
# open to everyone, same as picking a user.
|
||||
return ListGroups()
|
||||
|
||||
|
||||
class CreateGroupRequest(BaseModel):
|
||||
Name: str
|
||||
|
||||
|
||||
@Application.post("/api/groups")
|
||||
def PostGroupsEndpoint(Body: CreateGroupRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
|
||||
try:
|
||||
CreateGroup(Body.Name)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Created": True}
|
||||
|
||||
|
||||
@Application.delete("/api/groups")
|
||||
def DeleteGroupsEndpoint(Name: str, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
|
||||
try:
|
||||
DeleteGroup(Name)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Deleted": True}
|
||||
|
||||
|
||||
class SetGroupMembersRequest(BaseModel):
|
||||
Members: list[str]
|
||||
|
||||
|
||||
@Application.put("/api/groups/members")
|
||||
def PutGroupMembersEndpoint(
|
||||
Name: str, Body: SetGroupMembersRequest, CurrentUser: dict = Depends(RequireAdmin)
|
||||
) -> dict:
|
||||
try:
|
||||
SetGroupMembers(Name, Body.Members)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Saved": True}
|
||||
|
||||
|
||||
# --- Locked files/directories -----------------------------------------------
|
||||
# Setting a lock is open to anyone who currently has access to the path —
|
||||
# not admin-only. If you can see it, you're trusted to decide who else can.
|
||||
# Admins additionally bypass every lock (see Locks.IsPathAccessible).
|
||||
|
||||
|
||||
@Application.get("/api/lock")
|
||||
def GetLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
return GetLockForPath(Root, Path) or {}
|
||||
|
||||
|
||||
class SetLockRequest(BaseModel):
|
||||
Users: list[str] = []
|
||||
Groups: list[str] = []
|
||||
|
||||
|
||||
@Application.put("/api/lock")
|
||||
def PutLockEndpoint(
|
||||
Root: str, Path: str, Body: SetLockRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
SetLock(Root, Path, Body.Users, Body.Groups)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return {"Locked": True}
|
||||
|
||||
|
||||
@Application.delete("/api/lock")
|
||||
def DeleteLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
RemoveLock(Root, Path)
|
||||
return {"Unlocked": True}
|
||||
|
||||
|
||||
@Application.get("/api/locks")
|
||||
def GetLocksEndpoint(CurrentUser: dict = Depends(RequireAdmin)) -> list[dict]:
|
||||
# The full list (across every root/path) stays admin-only — it's a
|
||||
# global overview, not a per-item action.
|
||||
return LoadLocks()
|
||||
|
||||
|
||||
# --- Files ----------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -198,12 +374,32 @@ def GetRootsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
|
||||
def GetTreeEndpoint(
|
||||
Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> list[dict]:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
ShowHidden = LoadSettings(CurrentUser["Username"]).get("ShowHiddenFolders", False)
|
||||
return ListDirectory(Root, Path, ShowHidden)
|
||||
Entries = ListDirectory(Root, Path, ShowHidden)
|
||||
except (ValueError, NotADirectoryError, FileNotFoundError) as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
Username = CurrentUser["Username"]
|
||||
UserGroups = CurrentUser.get("Groups", [])
|
||||
IsAdmin = CurrentUser.get("IsAdmin", False)
|
||||
|
||||
# A locked subdirectory (or anything inside one) shouldn't even appear
|
||||
# in its parent's listing for someone without access to it.
|
||||
return [
|
||||
Entry
|
||||
for Entry in Entries
|
||||
if IsPathAccessible(
|
||||
Root,
|
||||
f"{Path}/{Entry['Name']}" if Path else Entry["Name"],
|
||||
Username,
|
||||
UserGroups,
|
||||
IsAdmin,
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
class CreateEntryRequest(BaseModel):
|
||||
IsDirectory: bool
|
||||
@@ -213,6 +409,8 @@ class CreateEntryRequest(BaseModel):
|
||||
def PostTreeEndpoint(
|
||||
Root: str, Path: str, Body: CreateEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
CreateEntry(Root, Path, Body.IsDirectory)
|
||||
except ValueError as Error:
|
||||
@@ -234,6 +432,8 @@ class MoveEntryRequest(BaseModel):
|
||||
def PatchTreeEndpoint(
|
||||
Root: str, Path: str, Body: MoveEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
if Body.DestinationPath:
|
||||
Destination = Body.DestinationPath
|
||||
elif Body.NewName:
|
||||
@@ -242,6 +442,8 @@ def PatchTreeEndpoint(
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="Provide NewName or DestinationPath.")
|
||||
|
||||
RequirePathAccessible(Root, Destination, CurrentUser)
|
||||
|
||||
try:
|
||||
MoveEntry(Root, Path, Destination)
|
||||
except ValueError as Error:
|
||||
@@ -258,6 +460,8 @@ def PatchTreeEndpoint(
|
||||
|
||||
@Application.delete("/api/tree")
|
||||
def DeleteTreeEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
DeleteEntry(Root, Path)
|
||||
except ValueError as Error:
|
||||
@@ -274,6 +478,8 @@ def DeleteTreeEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurr
|
||||
def GetFileEndpoint(
|
||||
Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
):
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
@@ -322,6 +528,8 @@ def GetFileEndpoint(
|
||||
async def PutFileEndpoint(
|
||||
Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
|
||||
):
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
@@ -348,12 +556,41 @@ async def PutFileEndpoint(
|
||||
|
||||
@Application.get("/api/blame")
|
||||
def GetBlameEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
raise HTTPException(status_code=400, detail=str(Error))
|
||||
|
||||
return LoadBlame(Root, Path)
|
||||
RawBlame = LoadBlame(Root, Path)
|
||||
AllUsers = {User["Username"]: User for User in ListUsers()}
|
||||
ViewerUsername = CurrentUser["Username"]
|
||||
Enriched: dict[str, dict] = {}
|
||||
|
||||
for LineNumber, Entry in RawBlame.items():
|
||||
AuthorUsername = Entry.get("Username")
|
||||
Timestamp = Entry.get("Timestamp")
|
||||
|
||||
if AuthorUsername == ViewerUsername:
|
||||
IsVisible = True
|
||||
elif AuthorUsername in AllUsers:
|
||||
IsVisible = LoadSettings(AuthorUsername).get("VisibleInBlame", True)
|
||||
else:
|
||||
IsVisible = True
|
||||
|
||||
if IsVisible:
|
||||
AuthorRecord = AllUsers.get(AuthorUsername, {})
|
||||
Enriched[LineNumber] = {
|
||||
"Username": AuthorUsername,
|
||||
"Timestamp": Timestamp,
|
||||
"Colour": AuthorRecord.get("Colour", "Purple"),
|
||||
"HasAvatar": AuthorUsername is not None and GetAvatarPath(AuthorUsername) is not None,
|
||||
}
|
||||
else:
|
||||
Enriched[LineNumber] = {"Username": None, "Hidden": True, "Timestamp": Timestamp}
|
||||
|
||||
return Enriched
|
||||
|
||||
|
||||
@Application.get("/api/raw/{RootName}/{FilePath:path}")
|
||||
@@ -365,6 +602,8 @@ def GetRawFileEndpoint(
|
||||
and scripts INSIDE a previewed HTML document resolve correctly against
|
||||
sibling files in the same directory.
|
||||
"""
|
||||
RequirePathAccessible(RootName, FilePath, CurrentUser)
|
||||
|
||||
try:
|
||||
ResolvedPath = ResolvePath(RootName, FilePath)
|
||||
except ValueError as Error:
|
||||
@@ -385,6 +624,8 @@ def GetRawFileEndpoint(
|
||||
|
||||
@Application.get("/api/book")
|
||||
def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
||||
RequirePathAccessible(Root, Path, CurrentUser)
|
||||
|
||||
try:
|
||||
FilePath = ResolvePath(Root, Path)
|
||||
except ValueError as Error:
|
||||
|
||||
@@ -42,6 +42,8 @@ DefaultSettings: dict[str, Any] = {
|
||||
"NewEntryButtonsAtTop": False,
|
||||
"StickyToolbar": True,
|
||||
"ConfirmBeforeDelete": True,
|
||||
"VisibleInBlame": True,
|
||||
"ShowBlameAvatars": False,
|
||||
}
|
||||
|
||||
DefaultThemesDirectory = ApplicationDirectory / "DefaultThemes"
|
||||
@@ -135,6 +137,13 @@ def SaveSettings(Username: str, NewSettings: dict[str, Any]) -> dict[str, Any]:
|
||||
return Merged
|
||||
|
||||
|
||||
def RenameUserSettings(OldUsername: str, NewUsername: str) -> None:
|
||||
OldPath = GetSettingsPath(OldUsername)
|
||||
|
||||
if OldPath.is_file():
|
||||
OldPath.rename(GetSettingsPath(NewUsername))
|
||||
|
||||
|
||||
def ListThemeNames() -> list[str]:
|
||||
EnsureDataDirectory()
|
||||
return sorted(ThemeFile.stem for ThemeFile in GetThemesDirectory().glob("*.toml"))
|
||||
|
||||
+54
-3
@@ -12,6 +12,7 @@ import tomli_w
|
||||
|
||||
from Settings import GetDataDirectory
|
||||
|
||||
from Groups import AddMember, AdminGroupName, GetGroupsForUser
|
||||
|
||||
Pbkdf2Iterations = 260_000
|
||||
|
||||
@@ -96,11 +97,13 @@ def CreateUser(Username: str, Password: str, IsAdmin: bool, Colour: str = Defaul
|
||||
|
||||
Users[Username] = {
|
||||
"PasswordHash": HashPassword(Password),
|
||||
"IsAdmin": IsAdmin,
|
||||
"Colour": Colour,
|
||||
}
|
||||
SaveUsers(Users)
|
||||
|
||||
if IsAdmin:
|
||||
AddMember(AdminGroupName, Username)
|
||||
|
||||
|
||||
def UpdateUserColour(Username: str, Colour: str) -> None:
|
||||
if Colour not in UserColourOptions:
|
||||
@@ -115,6 +118,44 @@ def UpdateUserColour(Username: str, Colour: str) -> None:
|
||||
SaveUsers(Users)
|
||||
|
||||
|
||||
def RenameUser(OldUsername: str, NewUsername: str) -> None:
|
||||
NewUsername = NewUsername.strip()
|
||||
|
||||
if not NewUsername:
|
||||
raise ValueError("Username cannot be empty.")
|
||||
|
||||
if NewUsername == OldUsername:
|
||||
return
|
||||
|
||||
Users = LoadUsers()
|
||||
|
||||
if OldUsername not in Users:
|
||||
raise ValueError("User not found.")
|
||||
|
||||
if NewUsername in Users:
|
||||
raise ValueError(f"User '{NewUsername}' already exists.")
|
||||
|
||||
Users[NewUsername] = Users.pop(OldUsername)
|
||||
SaveUsers(Users)
|
||||
|
||||
|
||||
def ChangePassword(Username: str, CurrentPassword: str, NewPassword: str) -> None:
|
||||
User = GetUser(Username)
|
||||
|
||||
if User is None:
|
||||
raise ValueError("User not found.")
|
||||
|
||||
if not VerifyPassword(CurrentPassword, User["PasswordHash"]):
|
||||
raise ValueError("Current password is incorrect.")
|
||||
|
||||
if not NewPassword:
|
||||
raise ValueError("New password cannot be empty.")
|
||||
|
||||
Users = LoadUsers()
|
||||
Users[Username]["PasswordHash"] = HashPassword(NewPassword)
|
||||
SaveUsers(Users)
|
||||
|
||||
|
||||
def VerifyLogin(Username: str, Password: str) -> dict[str, Any] | None:
|
||||
User = GetUser(Username)
|
||||
|
||||
@@ -126,7 +167,7 @@ def VerifyLogin(Username: str, Password: str) -> dict[str, Any] | None:
|
||||
|
||||
return {
|
||||
"Username": Username,
|
||||
"IsAdmin": User.get("IsAdmin", False),
|
||||
"IsAdmin": AdminGroupName in GetGroupsForUser(Username),
|
||||
"Colour": User.get("Colour", DefaultUserColour),
|
||||
}
|
||||
|
||||
@@ -136,7 +177,7 @@ def ListUsers() -> list[dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
"Username": Name,
|
||||
"IsAdmin": Data.get("IsAdmin", False),
|
||||
"IsAdmin": AdminGroupName in GetGroupsForUser(Name),
|
||||
"Colour": Data.get("Colour", DefaultUserColour),
|
||||
}
|
||||
for Name, Data in Users.items()
|
||||
@@ -168,3 +209,13 @@ def GetAvatarPath(Username: str) -> Path | None:
|
||||
|
||||
Matches = sorted(Directory.glob(f"{Username}.*"))
|
||||
return Matches[0] if Matches else None
|
||||
|
||||
|
||||
def RenameUserAvatar(OldUsername: str, NewUsername: str) -> None:
|
||||
Directory = GetAvatarDirectory()
|
||||
|
||||
if not Directory.is_dir():
|
||||
return
|
||||
|
||||
for ExistingPath in Directory.glob(f"{OldUsername}.*"):
|
||||
ExistingPath.rename(Directory / f"{NewUsername}{ExistingPath.suffix}")
|
||||
Reference in new issue
Block a user