Update.

Browse Source
This commit is contained in:
retoor 2025-11-09 01:34:53 +01:00
parent 9e9907bc00
commit 88d57c3837
8 changed files with 221 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
retoors/helpers/auth.py
View File

@ -6,13 +6,17 @@ from ..services.user_service import UserService # Import UserService
def login_required(func): def login_required(func):
@wraps(func) @wraps(func)
async def wrapper(self, *args, **kwargs): async def wrapper(self, *args, **kwargs):
session = await get_session(self.request) if not getattr(self, 'request', None):
request = self
else:
request = self.request
session = await get_session(request)
user_email = session.get('user_email') user_email = session.get('user_email')
if not user_email: if not user_email:
raise web.HTTPFound('/login') raise web.HTTPFound('/login')
user_service: UserService = self.request.app["user_service"] user_service: UserService = request.app["user_service"]
user = user_service.get_user_by_email(user_email) user = user_service.get_user_by_email(user_email)
if not user: if not user:
@ -21,6 +25,6 @@ def login_required(func):
raise web.HTTPFound('/login') raise web.HTTPFound('/login')
# Ensure the user object is available in the request for views # Ensure the user object is available in the request for views
self.request["user"] = user request["user"] = user
return await func(self, *args, **kwargs) return await func(self, *args, **kwargs)
return wrapper return wrapper

1
retoors/middlewares.py
View File

@ -20,3 +20,4 @@ async def error_middleware(request, handler):
raise # Re-raise HTTPException to see original traceback raise # Re-raise HTTPException to see original traceback
except Exception: except Exception:
raise # Re-raise generic Exception to see original traceback raise # Re-raise generic Exception to see original traceback

4
retoors/routes.py
View File

@ -32,9 +32,11 @@ def setup_routes(app):
app.router.add_view("/files", FileBrowserView, name="file_browser") app.router.add_view("/files", FileBrowserView, name="file_browser")
app.router.add_post("/files/new_folder", FileBrowserView, name="new_folder") app.router.add_post("/files/new_folder", FileBrowserView, name="new_folder")
app.router.add_post("/files/upload", UploadView, name="upload_file") app.router.add_post("/files/upload", UploadView, name="upload_file")
app.router.add_get("/files/download/{file_path:.*}", FileBrowserView, name="download_file") app.router.add_get("/files/download/{file_path:.*}", FileBrowserView.get_download_file, name="download_file")
app.router.add_post("/files/share/{file_path:.*}", FileBrowserView, name="share_file") app.router.add_post("/files/share/{file_path:.*}", FileBrowserView, name="share_file")
app.router.add_post("/files/delete/{file_path:.*}", FileBrowserView, name="delete_item") app.router.add_post("/files/delete/{file_path:.*}", FileBrowserView, name="delete_item")
app.router.add_get("/shared_file/{share_id}", FileBrowserView.shared_file_handler, name="shared_file")
app.router.add_get("/shared_file/{share_id}/download", FileBrowserView.download_shared_file_handler, name="download_shared_file")
# Admin API routes for user and team management # Admin API routes for user and team management
app.router.add_get("/api/users", get_users, name="api_get_users") app.router.add_get("/api/users", get_users, name="api_get_users")

68
retoors/services/file_service.py
View File

@ -4,38 +4,53 @@ from pathlib import Path
import shutil import shutil
import uuid import uuid
import datetime import datetime
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
class FileService: class FileService:
def __init__(self, base_dir: Path, users_data_path: Path): def __init__(self, base_dir: Path, users_data_path: Path):
self.base_dir = base_dir self.base_dir = base_dir
self.users_data_path = users_data_path self.users_data_path = users_data_path
self.base_dir.mkdir(parents=True, exist_ok=True) self.base_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"FileService initialized with base_dir: {self.base_dir} and users_data_path: {self.users_data_path}")
async def _load_users_data(self): async def _load_users_data(self):
"""Loads user data from the JSON file.""" """Loads user data from the JSON file."""
if not self.users_data_path.exists(): if not self.users_data_path.exists():
logger.warning(f"users_data_path does not exist: {self.users_data_path}")
return [] return []
async with aiofiles.open(self.users_data_path, mode="r") as f: async with aiofiles.open(self.users_data_path, mode="r") as f:
content = await f.read() content = await f.read()
try: try:
return json.loads(content) if content else [] return json.loads(content) if content else []
except json.JSONDecodeError: except json.JSONDecodeError:
logger.error(f"JSONDecodeError when loading users data from {self.users_data_path}")
return [] return []
async def _save_users_data(self, data): async def _save_users_data(self, data):
"""Saves user data to the JSON file.""" """Saves user data to the JSON file."""
async with aiofiles.open(self.users_data_path, mode="w") as f: async with aiofiles.open(self.users_data_path, mode="w") as f:
await f.write(json.dumps(data, indent=4)) await f.write(json.dumps(data, indent=4))
logger.debug(f"Saved users data to {self.users_data_path}")
def _get_user_file_path(self, user_email: str, relative_path: str = "") -> Path: def _get_user_file_path(self, user_email: str, relative_path: str = "") -> Path:
"""Constructs the absolute path for a user's file or directory.""" """Constructs the absolute path for a user's file or directory."""
user_dir = self.base_dir / user_email user_dir = self.base_dir / user_email
return user_dir / relative_path full_path = user_dir / relative_path
logger.debug(f"Constructed path for user '{user_email}', relative_path '{relative_path}': {full_path}")
return full_path
async def list_files(self, user_email: str, path: str = "") -> list: async def list_files(self, user_email: str, path: str = "") -> list:
"""Lists files and directories for a given user within a specified path.""" """Lists files and directories for a given user within a specified path."""
user_path = self._get_user_file_path(user_email, path) user_path = self._get_user_file_path(user_email, path)
if not user_path.is_dir(): if not user_path.is_dir():
logger.warning(f"list_files: User path is not a directory or does not exist: {user_path}")
return [] return []
files_list = [] files_list = []
@ -50,14 +65,17 @@ class FileService:
"last_modified": datetime.datetime.fromtimestamp(item.stat().st_mtime).isoformat(), "last_modified": datetime.datetime.fromtimestamp(item.stat().st_mtime).isoformat(),
} }
files_list.append(file_info) files_list.append(file_info)
logger.debug(f"Listed {len(files_list)} items for user '{user_email}' in path '{path}'")
return sorted(files_list, key=lambda x: (not x["is_dir"], x["name"].lower())) return sorted(files_list, key=lambda x: (not x["is_dir"], x["name"].lower()))
async def create_folder(self, user_email: str, folder_path: str) -> bool: async def create_folder(self, user_email: str, folder_path: str) -> bool:
"""Creates a new folder for the user.""" """Creates a new folder for the user."""
full_path = self._get_user_file_path(user_email, folder_path) full_path = self._get_user_file_path(user_email, folder_path)
if full_path.exists(): if full_path.exists():
logger.warning(f"create_folder: Folder already exists: {full_path}")
return False # Folder already exists return False # Folder already exists
full_path.mkdir(parents=True, exist_ok=True) full_path.mkdir(parents=True, exist_ok=True)
logger.info(f"create_folder: Folder created: {full_path}")
return True return True
async def upload_file(self, user_email: str, file_path: str, content: bytes) -> bool: async def upload_file(self, user_email: str, file_path: str, content: bytes) -> bool:
@ -66,38 +84,49 @@ class FileService:
full_path.parent.mkdir(parents=True, exist_ok=True) # Ensure parent directories exist full_path.parent.mkdir(parents=True, exist_ok=True) # Ensure parent directories exist
async with aiofiles.open(full_path, mode="wb") as f: async with aiofiles.open(full_path, mode="wb") as f:
await f.write(content) await f.write(content)
logger.info(f"upload_file: File uploaded to: {full_path}")
return True return True
async def download_file(self, user_email: str, file_path: str) -> tuple[bytes, str] | None: async def download_file(self, user_email: str, file_path: str) -> tuple[bytes, str] | None:
"""Downloads a file for the user.""" """Downloads a file for the user."""
full_path = self._get_user_file_path(user_email, file_path) full_path = self._get_user_file_path(user_email, file_path)
logger.debug(f"download_file: Attempting to download file from: {full_path}")
if full_path.is_file(): if full_path.is_file():
async with aiofiles.open(full_path, mode="rb") as f: async with aiofiles.open(full_path, mode="rb") as f:
content = await f.read() content = await f.read()
logger.info(f"download_file: Successfully read file: {full_path}")
return content, full_path.name return content, full_path.name
logger.warning(f"download_file: File not found or is not a file: {full_path}")
return None return None
async def delete_item(self, user_email: str, item_path: str) -> bool: async def delete_item(self, user_email: str, item_path: str) -> bool:
"""Deletes a file or folder for the user.""" """Deletes a file or folder for the user."""
full_path = self._get_user_file_path(user_email, item_path) full_path = self._get_user_file_path(user_email, item_path)
logger.debug(f"delete_item: Attempting to delete item: {full_path}")
if not full_path.exists(): if not full_path.exists():
logger.warning(f"delete_item: Item does not exist: {full_path}")
return False return False
if full_path.is_file(): if full_path.is_file():
full_path.unlink() full_path.unlink()
logger.info(f"delete_item: File deleted: {full_path}")
elif full_path.is_dir(): elif full_path.is_dir():
shutil.rmtree(full_path) shutil.rmtree(full_path)
logger.info(f"delete_item: Directory deleted: {full_path}")
return True return True
async def generate_share_link(self, user_email: str, item_path: str) -> str | None: async def generate_share_link(self, user_email: str, item_path: str) -> str | None:
"""Generates a shareable link for a file or folder.""" """Generates a shareable link for a file or folder."""
logger.debug(f"generate_share_link: Generating link for user '{user_email}', item '{item_path}'")
users_data = await self._load_users_data() users_data = await self._load_users_data()
user = next((u for u in users_data if u.get("email") == user_email), None) user = next((u for u in users_data if u.get("email") == user_email), None)
if not user: if not user:
logger.warning(f"generate_share_link: User not found: {user_email}")
return None return None
full_path = self._get_user_file_path(user_email, item_path) full_path = self._get_user_file_path(user_email, item_path)
if not full_path.exists(): if not full_path.exists():
logger.warning(f"generate_share_link: Item does not exist: {full_path}")
return None return None
share_id = str(uuid.uuid4()) share_id = str(uuid.uuid4())
@ -110,37 +139,60 @@ class FileService:
"expires_at": (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=7)).isoformat(), # 7-day expiry "expires_at": (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=7)).isoformat(), # 7-day expiry
} }
await self._save_users_data(users_data) await self._save_users_data(users_data)
logger.info(f"generate_share_link: Share link generated with ID: {share_id} for item: {item_path}")
return share_id return share_id
async def get_shared_item(self, share_id: str) -> dict | None: async def get_shared_item(self, share_id: str) -> dict | None:
"""Retrieves information about a shared item.""" """Retrieves information about a shared item."""
logger.debug(f"get_shared_item: Retrieving shared item with ID: {share_id}")
users_data = await self._load_users_data() users_data = await self._load_users_data()
for user_info in users_data: for user_info in users_data:
if "shared_items" in user_info and share_id in user_info["shared_items"]: if "shared_items" in user_info and share_id in user_info["shared_items"]:
shared_item = user_info["shared_items"][share_id] shared_item = user_info["shared_items"][share_id]
expiry_time = datetime.datetime.fromisoformat(shared_item["expires_at"]) expiry_time = datetime.datetime.fromisoformat(shared_item["expires_at"])
if expiry_time > datetime.datetime.now(datetime.timezone.utc): if expiry_time > datetime.datetime.now(datetime.timezone.utc):
logger.info(f"get_shared_item: Found valid shared item for ID: {share_id}")
return shared_item return shared_item
else:
logger.warning(f"get_shared_item: Shared item {share_id} has expired.")
logger.warning(f"get_shared_item: No valid shared item found for ID: {share_id}")
return None return None
async def get_shared_file_content(self, share_id: str) -> tuple[bytes, str] | None: async def get_shared_file_content(self, share_id: str, requested_file_path: str | None = None) -> tuple[bytes, str] | None:
"""Retrieves the content of a shared file.""" """Retrieves the content of a shared file."""
logger.debug(f"get_shared_file_content: Retrieving content for shared file with ID: {share_id}, requested_file_path: {requested_file_path}")
shared_item = await self.get_shared_item(share_id) shared_item = await self.get_shared_item(share_id)
if not shared_item: if not shared_item:
return None return None
user_email = shared_item["user_email"] user_email = shared_item["user_email"]
item_path = shared_item["item_path"] item_path = shared_item["item_path"] # This is the path of the originally shared item (file or folder)
full_path = self._get_user_file_path(user_email, item_path)
if full_path.is_file(): # Construct the full path to the originally shared item
async with aiofiles.open(full_path, mode="rb") as f: full_shared_item_path = self._get_user_file_path(user_email, item_path)
target_file_path = full_shared_item_path
if requested_file_path:
# If a specific file within a shared folder is requested
target_file_path = self._get_user_file_path(user_email, requested_file_path)
# Security check: Ensure the requested file is actually within the shared item's directory
try:
target_file_path.relative_to(full_shared_item_path)
except ValueError:
logger.warning(f"get_shared_file_content: Requested file path '{requested_file_path}' is not within shared item path '{item_path}' for share_id: {share_id}")
return None
if target_file_path.is_file():
async with aiofiles.open(target_file_path, mode="rb") as f:
content = await f.read() content = await f.read()
return content, full_path.name logger.info(f"get_shared_file_content: Successfully read content for shared file: {target_file_path}")
return content, target_file_path.name
logger.warning(f"get_shared_file_content: Shared item path is not a file or does not exist: {target_file_path}")
return None return None
async def get_shared_folder_content(self, share_id: str) -> list | None: async def get_shared_folder_content(self, share_id: str) -> list | None:
"""Retrieves the content of a shared folder.""" """Retrieves the content of a shared folder."""
logger.debug(f"get_shared_folder_content: Retrieving content for shared folder with ID: {share_id}")
shared_item = await self.get_shared_item(share_id) shared_item = await self.get_shared_item(share_id)
if not shared_item: if not shared_item:
return None return None
@ -150,5 +202,7 @@ class FileService:
full_path = self._get_user_file_path(user_email, item_path) full_path = self._get_user_file_path(user_email, item_path)
if full_path.is_dir(): if full_path.is_dir():
logger.info(f"get_shared_folder_content: Listing files for shared folder: {full_path}")
return await self.list_files(user_email, item_path) return await self.list_files(user_email, item_path)
logger.warning(f"get_shared_folder_content: Shared item path is not a directory or does not exist: {full_path}")
return None return None

4
retoors/static/css/components/file_browser.css
View File

@ -65,8 +65,8 @@
} }
.btn-small { .btn-small {
padding: 0.4rem 0.8rem; padding: 0.2rem 0.4rem; /* Reduced padding */
font-size: 0.85rem; font-size: 0.75rem; /* Reduced font size */
border: 1px solid var(--border-color); border: 1px solid var(--border-color);
border-radius: 4px; border-radius: 4px;
background-color: var(--background-color); background-color: var(--background-color);

8
retoors/templates/pages/file_browser.html
View File

@ -11,8 +11,8 @@
{% block dashboard_actions %} {% block dashboard_actions %}
<button class="btn-primary" id="new-folder-btn">+ New</button> <button class="btn-primary" id="new-folder-btn">+ New</button>
<button class="btn-outline" id="upload-btn">Upload</button> <button class="btn-outline" id="upload-btn">Upload</button>
<button class="btn-outline" id="download-selected-btn" disabled>Download</button> <button class="btn-outline" id="download-selected-btn" disabled>⬇️</button>
<button class="btn-outline" id="share-selected-btn" disabled>Share</button> <button class="btn-outline" id="share-selected-btn" disabled>🔗</button>
<button class="btn-outline" id="delete-selected-btn" disabled>Delete</button> <button class="btn-outline" id="delete-selected-btn" disabled>Delete</button>
{% endblock %} {% endblock %}
@ -70,9 +70,9 @@
<td> <td>
<div class="action-buttons"> <div class="action-buttons">
{% if not item.is_dir %} {% if not item.is_dir %}
<button class="btn-small download-file-btn" data-path="{{ item.path }}">Download</button> <button class="btn-small download-file-btn" data-path="{{ item.path }}">⬇️</button>
{% endif %} {% endif %}
<button class="btn-small share-file-btn" data-path="{{ item.path }}" data-name="{{ item.name }}">Share</button> <button class="btn-small share-file-btn" data-path="{{ item.path }}" data-name="{{ item.name }}">🔗</button>
<button class="btn-small btn-danger delete-file-btn" data-path="{{ item.path }}" data-name="{{ item.name }}">Delete</button> <button class="btn-small btn-danger delete-file-btn" data-path="{{ item.path }}" data-name="{{ item.name }}">Delete</button>
</div> </div>
</td> </td>

166
retoors/views/site.py
View File

@ -1,3 +1,4 @@
import logging
from aiohttp import web from aiohttp import web
import aiohttp_jinja2 import aiohttp_jinja2
import os import os
@ -7,6 +8,14 @@ from aiohttp.web_response import json_response
from ..helpers.auth import login_required from ..helpers.auth import login_required
from .auth import CustomPydanticView from .auth import CustomPydanticView
# Configure logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
# PROJECT_DIR is no longer directly used for user files, as FileService manages them # PROJECT_DIR is no longer directly used for user files, as FileService manages them
# PROJECT_DIR = Path(__file__).parent.parent.parent / "project" # PROJECT_DIR = Path(__file__).parent.parent.parent / "project"
@ -45,7 +54,7 @@ class SiteView(web.View):
@login_required @login_required
async def dashboard(self): async def dashboard(self):
return web.HTTPFound(self.request.app.router["file_browser"].url_for()) raise web.HTTPFound(self.request.app.router["file_browser"].url_for())
async def solutions(self): async def solutions(self):
return aiohttp_jinja2.render_template( return aiohttp_jinja2.render_template(
@ -126,9 +135,6 @@ class SiteView(web.View):
class FileBrowserView(web.View): class FileBrowserView(web.View):
@login_required @login_required
async def get(self): async def get(self):
if self.request.match_info.get("file_path") is not None:
return await self.get_download_file()
user_email = self.request["user"]["email"] user_email = self.request["user"]["email"]
file_service = self.request.app["file_service"] file_service = self.request.app["file_service"]
@ -157,12 +163,14 @@ class FileBrowserView(web.View):
user_email = self.request["user"]["email"] user_email = self.request["user"]["email"]
file_service = self.request.app["file_service"] file_service = self.request.app["file_service"]
route_name = self.request.match_info.route.name route_name = self.request.match_info.route.name
logger.debug(f"FileBrowserView: POST request for route: {route_name}")
if route_name == "new_folder": if route_name == "new_folder":
data = await self.request.post() data = await self.request.post()
folder_name = data.get("folder_name") folder_name = data.get("folder_name")
if not folder_name: if not folder_name:
return web.HTTPFound( logger.warning("FileBrowserView: New folder request missing folder_name")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
error="Folder name is required" error="Folder name is required"
) )
@ -170,13 +178,15 @@ class FileBrowserView(web.View):
success = await file_service.create_folder(user_email, folder_name) success = await file_service.create_folder(user_email, folder_name)
if success: if success:
return web.HTTPFound( logger.info(f"FileBrowserView: Folder '{folder_name}' created successfully for user {user_email}")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
success=f"Folder '{folder_name}' created successfully" success=f"Folder '{folder_name}' created successfully"
) )
) )
else: else:
return web.HTTPFound( logger.error(f"FileBrowserView: Failed to create folder '{folder_name}' for user {user_email}")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
error=f"Folder '{folder_name}' already exists or could not be created" error=f"Folder '{folder_name}' already exists or could not be created"
) )
@ -184,20 +194,26 @@ class FileBrowserView(web.View):
elif route_name == "share_file": elif route_name == "share_file":
file_path = self.request.match_info.get("file_path") file_path = self.request.match_info.get("file_path")
logger.debug(f"FileBrowserView: Share file request for path: {file_path} by user {user_email}")
if not file_path: if not file_path:
logger.warning("FileBrowserView: Share file request missing file_path")
return json_response({"error": "File path is required for sharing"}, status=400) return json_response({"error": "File path is required for sharing"}, status=400)
share_id = await file_service.generate_share_link(user_email, file_path) share_id = await file_service.generate_share_link(user_email, file_path)
if share_id: if share_id:
share_link = f"{self.request.scheme}://{self.request.host}/shared_file/{share_id}" share_link = f"{self.request.scheme}://{self.request.host}/shared_file/{share_id}"
logger.info(f"FileBrowserView: Share link generated: {share_link} for file: {file_path}")
return json_response({"share_link": share_link}) return json_response({"share_link": share_link})
else: else:
logger.error(f"FileBrowserView: Failed to generate share link for file: {file_path} by user {user_email}")
return json_response({"error": "Failed to generate share link"}, status=500) return json_response({"error": "Failed to generate share link"}, status=500)
elif route_name == "delete_item": elif route_name == "delete_item":
item_path = self.request.match_info.get("file_path") item_path = self.request.match_info.get("file_path")
logger.debug(f"FileBrowserView: Delete item request for path: {item_path} by user {user_email}")
if not item_path: if not item_path:
return web.HTTPFound( logger.warning("FileBrowserView: Delete item request missing item_path")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
error="Item path is required for deletion" error="Item path is required for deletion"
) )
@ -205,38 +221,134 @@ class FileBrowserView(web.View):
success = await file_service.delete_item(user_email, item_path) success = await file_service.delete_item(user_email, item_path)
if success: if success:
return web.HTTPFound( logger.info(f"FileBrowserView: Item '{item_path}' deleted successfully for user {user_email}")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
success=f"Item deleted successfully" success=f"Item deleted successfully"
) )
) )
else: else:
return web.HTTPFound( logger.error(f"FileBrowserView: Failed to delete item '{item_path}' for user {user_email}")
raise web.HTTPFound(
self.request.app.router["file_browser"].url_for().with_query( self.request.app.router["file_browser"].url_for().with_query(
error="Failed to delete item - it may not exist" error="Failed to delete item - it may not exist"
) )
) )
return web.HTTPBadRequest(text="Unknown file action") logger.warning(f"FileBrowserView: Unknown file action for POST request: {route_name}")
raise web.HTTPBadRequest(text="Unknown file action")
@login_required @login_required
async def get_download_file(self): async def get_download_file(request):
user_email = self.request["user"]["email"] user_email = request["user"]["email"]
file_service = self.request.app["file_service"] file_service = request.app["file_service"]
file_path = self.request.match_info.get("file_path") file_path = request.match_info.get("file_path")
logger.debug(f"FileBrowserView: Download file request for path: {file_path} by user {user_email}")
if not file_path: if not file_path:
return web.HTTPBadRequest(text="File path is required for download") logger.warning("FileBrowserView: Download file request missing file_path")
raise web.HTTPBadRequest(text="File path is required for download")
result = await file_service.download_file(user_email, file_path) result = await file_service.download_file(user_email, file_path)
if result: if result:
content, filename = result content, filename = result
response = web.Response(body=content) response = web.Response(body=content)
response.headers["Content-Disposition"] = f"attachment; filename={filename}" response.headers["Content-Disposition"] = f"attachment; filename={filename}"
response.headers["Content-Type"] = "application/octet-stream" response.headers["Content-Type"] = "application/octet-stream"
logger.info(f"FileBrowserView: File '{filename}' downloaded successfully by user {user_email}")
return response return response
else: else:
return web.HTTPNotFound(text="File not found") logger.error(f"FileBrowserView: Failed to download file: {file_path} for user {user_email} - file not found or access denied")
raise web.HTTPNotFound(text="File not found")
async def shared_file_handler(self):
share_id = self.request.match_info.get("share_id")
file_service = self.request.app["file_service"]
logger.debug(f"FileBrowserView: Handling shared file request for share_id: {share_id}")
shared_item = await file_service.get_shared_item(share_id)
if not shared_item:
logger.warning(f"FileBrowserView: Shared item not found or expired for share_id: {share_id}")
return aiohttp_jinja2.render_template(
"pages/errors/404.html",
self.request,
{"request": self.request, "message": "Shared link is invalid or has expired."}
)
user_email = shared_item["user_email"]
item_path = shared_item["item_path"]
full_path = file_service._get_user_file_path(user_email, item_path)
if full_path.is_file():
result = await file_service.get_shared_file_content(share_id)
if result:
content, filename = result
response = web.Response(body=content)
response.headers["Content-Disposition"] = f"attachment; filename={filename}"
response.headers["Content-Type"] = "application/octet-stream"
logger.info(f"FileBrowserView: Serving shared file '{filename}' for share_id: {share_id}")
return response
else:
logger.error(f"FileBrowserView: Failed to get content for shared file: {item_path} (share_id: {share_id})")
raise web.HTTPNotFound(text="Shared file not found or inaccessible")
elif full_path.is_dir():
files = await file_service.get_shared_folder_content(share_id)
logger.info(f"FileBrowserView: Serving shared folder '{item_path}' for share_id: {share_id}")
return aiohttp_jinja2.render_template(
"pages/shared_folder.html",
self.request,
{
"request": self.request,
"files": files,
"current_path": item_path,
"share_id": share_id,
"user_email": user_email,
"active_page": "shared"
}
)
else:
logger.error(f"FileBrowserView: Shared item is neither file nor directory: {item_path} (share_id: {share_id})")
raise web.HTTPNotFound(text="Shared item not found")
async def download_shared_file_handler(self):
share_id = self.request.match_info.get("share_id")
file_path = self.request.query.get("file_path") # This is the path of the file *within* the shared item
file_service = self.request.app["file_service"]
logger.debug(f"FileBrowserView: Handling download shared file request for share_id: {share_id}, file_path: {file_path}")
if not file_path:
logger.warning("FileBrowserView: Download shared file request missing file_path query parameter.")
raise web.HTTPBadRequest(text="File path is required for download from shared folder.")
shared_item = await file_service.get_shared_item(share_id)
if not shared_item:
logger.warning(f"FileBrowserView: Shared item not found or expired for share_id: {share_id} during download.")
return aiohttp_jinja2.render_template(
"pages/errors/404.html",
self.request,
{"request": self.request, "message": "Shared link is invalid or has expired."}
)
# Ensure the shared item is a directory if a file_path is provided
user_email = shared_item["user_email"]
original_shared_item_path = file_service._get_user_file_path(user_email, shared_item["item_path"])
if not original_shared_item_path.is_dir():
logger.warning(f"FileBrowserView: Attempt to download a specific file from a shared item that is not a directory. Share_id: {share_id}")
raise web.HTTPBadRequest(text="Cannot download specific files from a shared item that is not a folder.")
result = await file_service.get_shared_file_content(share_id, file_path)
if result:
content, filename = result
response = web.Response(body=content)
response.headers["Content-Disposition"] = f"attachment; filename={filename}"
response.headers["Content-Type"] = "application/octet-stream"
logger.info(f"FileBrowserView: Serving shared file '{filename}' from shared folder for share_id: {share_id}")
return response
else:
logger.error(f"FileBrowserView: Failed to get content for shared file: {file_path} (share_id: {share_id}) from shared folder.")
raise web.HTTPNotFound(text="Shared file not found or inaccessible within the shared folder.")
class OrderView(CustomPydanticView): class OrderView(CustomPydanticView):
@ -283,7 +395,7 @@ class UserManagementView(web.View):
elif route_name == "user_details": elif route_name == "user_details":
return await self.user_details_page() return await self.user_details_page()
return web.HTTPNotFound() raise web.HTTPNotFound()
@login_required @login_required
async def post(self): async def post(self):
@ -296,7 +408,7 @@ class UserManagementView(web.View):
elif route_name == "delete_user_page": elif route_name == "delete_user_page":
return await self.delete_user_submit() return await self.delete_user_submit()
return web.HTTPNotFound() raise web.HTTPNotFound()
async def add_user_page(self): async def add_user_page(self):
return aiohttp_jinja2.render_template( return aiohttp_jinja2.render_template(
@ -363,7 +475,7 @@ class UserManagementView(web.View):
user_service.update_user_quota(email, float(storage_quota_gb)) user_service.update_user_quota(email, float(storage_quota_gb))
return web.HTTPFound( raise web.HTTPFound(
self.request.app.router["users"].url_for().with_query( self.request.app.router["users"].url_for().with_query(
success=f"User {email} added successfully" success=f"User {email} added successfully"
) )
@ -389,7 +501,7 @@ class UserManagementView(web.View):
user_data = user_service.get_user_by_email(email) user_data = user_service.get_user_by_email(email)
if not user_data: if not user_data:
return web.HTTPNotFound(text="User not found") raise web.HTTPNotFound(text="User not found")
success_message = self.request.query.get("success") success_message = self.request.query.get("success")
@ -424,7 +536,7 @@ class UserManagementView(web.View):
user_data = user_service.get_user_by_email(email) user_data = user_service.get_user_by_email(email)
if not user_data: if not user_data:
return web.HTTPNotFound(text="User not found") raise web.HTTPNotFound(text="User not found")
if errors: if errors:
return aiohttp_jinja2.render_template( return aiohttp_jinja2.render_template(
@ -441,7 +553,7 @@ class UserManagementView(web.View):
user_service.update_user_quota(email, storage_quota_gb) user_service.update_user_quota(email, storage_quota_gb)
return web.HTTPFound( raise web.HTTPFound(
self.request.app.router["edit_user"].url_for(email=email).with_query( self.request.app.router["edit_user"].url_for(email=email).with_query(
success="User quota updated successfully" success="User quota updated successfully"
) )
@ -454,7 +566,7 @@ class UserManagementView(web.View):
user_data = user_service.get_user_by_email(email) user_data = user_service.get_user_by_email(email)
if not user_data: if not user_data:
return web.HTTPNotFound(text="User not found") raise web.HTTPNotFound(text="User not found")
return aiohttp_jinja2.render_template( return aiohttp_jinja2.render_template(
"pages/user_details.html", "pages/user_details.html",
@ -474,11 +586,11 @@ class UserManagementView(web.View):
user_data = user_service.get_user_by_email(email) user_data = user_service.get_user_by_email(email)
if not user_data: if not user_data:
return web.HTTPNotFound(text="User not found") raise web.HTTPNotFound(text="User not found")
user_service.delete_user(email) user_service.delete_user(email)
return web.HTTPFound( raise web.HTTPFound(
self.request.app.router["users"].url_for().with_query( self.request.app.router["users"].url_for().with_query(
success=f"User {email} deleted successfully" success=f"User {email} deleted successfully"
) )

6
tests/test_file_browser.py
View File

@ -4,6 +4,7 @@ from pathlib import Path
import json import json
import datetime import datetime
import aiohttp import aiohttp
from aiohttp import web
from aiohttp.test_utils import TestClient from aiohttp.test_utils import TestClient
from aiohttp_session import setup as setup_session from aiohttp_session import setup as setup_session
from aiohttp_session.cookie_storage import EncryptedCookieStorage from aiohttp_session.cookie_storage import EncryptedCookieStorage
@ -389,8 +390,9 @@ async def test_file_browser_download_file(logged_in_client: TestClient, file_ser
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_file_browser_download_file_not_found(logged_in_client: TestClient): async def test_file_browser_download_file_not_found(logged_in_client: TestClient):
resp = await logged_in_client.get("/files/download/nonexistent_web.txt", allow_redirects=False) response = await logged_in_client.get("/files/download/nonexistent_web.txt", allow_redirects=False)
assert resp.status == 404 assert response.status == 404
assert "File not found" in await response.text()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_file_browser_delete_file(logged_in_client: TestClient, file_service_instance, temp_user_files_dir): async def test_file_browser_delete_file(logged_in_client: TestClient, file_service_instance, temp_user_files_dir):