feat(sharing): add file sharing and role-based access management

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/8091889d-4810-4794-b9d3-6b8f7f5257c4
This commit is contained in:
copilot-swe-agent[bot]
2026-03-22 14:14:50 +00:00
parent 83afc6c8f6
commit 6f2752bdf8
13 changed files with 1669 additions and 9 deletions
+2
View File
@@ -46,6 +46,7 @@ from app.api.sessions import router as sessions_router
from app.api.settings import router as settings_router from app.api.settings import router as settings_router
from app.api.shared_links import public_router as shared_links_public_router from app.api.shared_links import public_router as shared_links_public_router
from app.api.shared_links import router as shared_links_router from app.api.shared_links import router as shared_links_router
from app.api.sharing import router as sharing_router
from app.api.similarity import router as similarity_router from app.api.similarity import router as similarity_router
from app.api.subscriptions import router as subscriptions_router from app.api.subscriptions import router as subscriptions_router
from app.api.system_reset import router as system_reset_router from app.api.system_reset import router as system_reset_router
@@ -110,3 +111,4 @@ router.include_router(translation_router)
router.include_router(classification_rules_router) router.include_router(classification_rules_router)
router.include_router(automation_router) router.include_router(automation_router)
router.include_router(comments_router) router.include_router(comments_router)
router.include_router(sharing_router)
+79 -3
View File
@@ -15,7 +15,15 @@ from sqlalchemy.orm import Session
from app.auth import get_current_user_id, require_login from app.auth import get_current_user_id, require_login
from app.database import get_db from app.database import get_db
from app.models import DocumentAnnotation, DocumentComment, FileRecord, UserProfile from app.models import (
FILE_SHARE_ROLE_VIEWER,
DocumentAnnotation,
DocumentComment,
FileRecord,
FileShare,
UserProfile,
)
from app.utils.user_scope import get_current_owner_id, has_file_role
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -141,7 +149,7 @@ def list_comments(request: Request, file_id: int, db: DbSession):
"""List all comments for a document, organized into threads. """List all comments for a document, organized into threads.
Returns a threaded tree where top-level comments contain nested Returns a threaded tree where top-level comments contain nested
``replies``. ``replies``. Requires at least viewer access.
Path Parameters: Path Parameters:
file_id: The ID of the document. file_id: The ID of the document.
@@ -149,10 +157,17 @@ def list_comments(request: Request, file_id: int, db: DbSession):
Returns: Returns:
A dict with ``file_id``, ``comments`` (threaded), and ``total``. A dict with ``file_id``, ``comments`` (threaded), and ``total``.
""" """
user_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record: if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not is_admin and not has_file_role(file_record, user_id, db, minimum_role=FILE_SHARE_ROLE_VIEWER):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
comments = ( comments = (
db.query(DocumentComment).filter(DocumentComment.file_id == file_id).order_by(DocumentComment.created_at).all() db.query(DocumentComment).filter(DocumentComment.file_id == file_id).order_by(DocumentComment.created_at).all()
) )
@@ -176,7 +191,10 @@ def create_comment(
"""Create a new comment on a document. """Create a new comment on a document.
Automatically extracts @mentions from the comment body and stores Automatically extracts @mentions from the comment body and stores
them for later notification or UI highlighting. them for later notification or UI highlighting. When multi-user
mode is enabled, any mentioned user that does not already have
access to the document is automatically granted ``viewer`` access by
the file owner so they can read the file and continue the discussion.
Path Parameters: Path Parameters:
file_id: The ID of the document to comment on. file_id: The ID of the document to comment on.
@@ -189,11 +207,17 @@ def create_comment(
The created comment object. The created comment object.
""" """
user_id = get_current_user_id(request) user_id = get_current_user_id(request)
owner_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record: if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not is_admin and not has_file_role(file_record, owner_id, db, minimum_role=FILE_SHARE_ROLE_VIEWER):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not isinstance(body, str) or not body.strip(): if not isinstance(body, str) or not body.strip():
raise HTTPException( raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -230,8 +254,45 @@ def create_comment(
try: try:
db.add(comment) db.add(comment)
db.flush() # write comment so we can get its id before committing
# Auto-share the file with mentioned users that don't have access yet.
# Only do this in multi-user mode and only when the file has an owner
# (unowned files are already visible to all authenticated users).
if mentions and file_record.owner_id is not None:
from app.config import settings as _settings
if _settings.multi_user_enabled:
for mentioned_user in mentions:
# Skip the commenter themselves and the file owner
if mentioned_user in (owner_id, file_record.owner_id):
continue
existing_share = (
db.query(FileShare)
.filter(
FileShare.file_id == file_id,
FileShare.shared_with_user_id == mentioned_user,
)
.first()
)
if not existing_share:
auto_share = FileShare(
file_id=file_id,
owner_id=file_record.owner_id,
shared_with_user_id=mentioned_user,
role=FILE_SHARE_ROLE_VIEWER,
)
db.add(auto_share)
logger.info(
"Auto-shared file_id=%s with mentioned user=%s as viewer",
file_id,
mentioned_user,
)
db.commit() db.commit()
db.refresh(comment) db.refresh(comment)
except HTTPException:
raise
except Exception: except Exception:
db.rollback() db.rollback()
logger.exception("Failed to create comment on file_id=%s", file_id) logger.exception("Failed to create comment on file_id=%s", file_id)
@@ -402,16 +463,25 @@ def resolve_comment(
def list_annotations(request: Request, file_id: int, db: DbSession): def list_annotations(request: Request, file_id: int, db: DbSession):
"""List all annotations for a document. """List all annotations for a document.
Requires at least viewer access.
Path Parameters: Path Parameters:
file_id: The ID of the document. file_id: The ID of the document.
Returns: Returns:
A dict with ``file_id``, ``annotations``, and ``total``. A dict with ``file_id``, ``annotations``, and ``total``.
""" """
user_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record: if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not is_admin and not has_file_role(file_record, user_id, db, minimum_role=FILE_SHARE_ROLE_VIEWER):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
annotations = ( annotations = (
db.query(DocumentAnnotation) db.query(DocumentAnnotation)
.filter(DocumentAnnotation.file_id == file_id) .filter(DocumentAnnotation.file_id == file_id)
@@ -461,11 +531,17 @@ def create_annotation(
The created annotation object. The created annotation object.
""" """
user_id = get_current_user_id(request) user_id = get_current_user_id(request)
owner_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record: if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not is_admin and not has_file_role(file_record, owner_id, db, minimum_role=FILE_SHARE_ROLE_VIEWER):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if not isinstance(content, str) or not content.strip(): if not isinstance(content, str) or not content.strip():
raise HTTPException( raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+27 -1
View File
@@ -30,7 +30,7 @@ from app.utils.file_queries import apply_status_filter
from app.utils.file_status import get_files_processing_status from app.utils.file_status import get_files_processing_status
from app.utils.filename_utils import sanitize_filename from app.utils.filename_utils import sanitize_filename
from app.utils.input_validation import validate_search_query, validate_sort_field, validate_sort_order from app.utils.input_validation import validate_search_query, validate_sort_field, validate_sort_order
from app.utils.user_scope import apply_owner_filter, get_current_owner_id from app.utils.user_scope import apply_owner_filter, get_current_owner_id, get_file_role
# Set up logging # Set up logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -300,6 +300,7 @@ def delete_file_record(request: Request, file_id: int, db: DbSession):
""" """
Delete a file record from the database. Delete a file record from the database.
This only removes the database entry, not the actual file. This only removes the database entry, not the actual file.
Only the file owner (or an admin) may delete a document.
""" """
# Check if file deletion is allowed # Check if file deletion is allowed
if not settings.allow_file_delete: if not settings.allow_file_delete:
@@ -314,6 +315,18 @@ def delete_file_record(request: Request, file_id: int, db: DbSession):
if not file_record: if not file_record:
raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found") raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")
# Enforce owner-only deletion in multi-user mode
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
if not is_admin:
owner_id = get_current_owner_id(request)
role = get_file_role(file_record, owner_id, db)
if role != "owner":
raise HTTPException(
status_code=403,
detail="Only the file owner can delete this document",
)
# Log the deletion # Log the deletion
logger.info(f"Deleting file record: ID={file_id}, Filename={file_record.original_filename}") logger.info(f"Deleting file record: ID={file_id}, Filename={file_record.original_filename}")
@@ -340,6 +353,7 @@ def bulk_delete_files(request: Request, file_ids: List[int], db: DbSession):
""" """
Delete multiple file records from the database. Delete multiple file records from the database.
This only removes the database entries, not the actual files. This only removes the database entries, not the actual files.
Only the file owner (or an admin) may delete each document.
""" """
# Check if file deletion is allowed # Check if file deletion is allowed
if not settings.allow_file_delete: if not settings.allow_file_delete:
@@ -354,6 +368,18 @@ def bulk_delete_files(request: Request, file_ids: List[int], db: DbSession):
if not file_records: if not file_records:
raise HTTPException(status_code=404, detail="No files found with the provided IDs") raise HTTPException(status_code=404, detail="No files found with the provided IDs")
# Enforce owner-only deletion in multi-user mode
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
if not is_admin:
owner_id = get_current_owner_id(request)
non_owner_ids = [f.id for f in file_records if get_file_role(f, owner_id, db) != "owner"]
if non_owner_ids:
raise HTTPException(
status_code=403,
detail=f"You can only delete files you own. Not owner of file IDs: {non_owner_ids}",
)
deleted_count = len(file_records) deleted_count = len(file_records)
deleted_ids = [f.id for f in file_records] deleted_ids = [f.id for f in file_records]
+355
View File
@@ -0,0 +1,355 @@
"""File-sharing API endpoints.
Provides CRUD operations for ``FileShare`` records, which grant named
users ``viewer`` or ``editor`` access to a document owned by someone
else. Only the file owner may create, update, or revoke shares.
"""
import logging
from typing import Annotated, Any
from fastapi import APIRouter, Body, Depends, HTTPException, Request, status
from sqlalchemy.orm import Session
from app.auth import require_login
from app.database import get_db
from app.models import FILE_SHARE_ROLE_VIEWER, FILE_SHARE_ROLES, FileRecord, FileShare, UserProfile
from app.utils.user_scope import get_current_owner_id, get_file_role
logger = logging.getLogger(__name__)
router = APIRouter(tags=["sharing"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _serialize_share(share: FileShare) -> dict[str, Any]:
"""Serialize a ``FileShare`` to a JSON-friendly dict."""
return {
"id": share.id,
"file_id": share.file_id,
"owner_id": share.owner_id,
"shared_with_user_id": share.shared_with_user_id,
"role": share.role,
"created_at": share.created_at.isoformat() if share.created_at else None,
"updated_at": share.updated_at.isoformat() if share.updated_at else None,
}
def _require_owner(file_record: FileRecord, user_id: str | None, db: Session) -> None:
"""Raise 403 unless the calling user is the file owner."""
if get_file_role(file_record, user_id, db) != "owner":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the file owner can manage shares",
)
# ---------------------------------------------------------------------------
# List shares
# ---------------------------------------------------------------------------
@router.get("/files/{file_id}/shares")
@require_login
def list_shares(request: Request, file_id: int, db: DbSession):
"""List all shares for a document.
Only the file owner (or an admin) may call this endpoint.
Path Parameters:
file_id: The ID of the document.
Returns:
A list of share objects.
"""
user_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
role = get_file_role(file_record, user_id, db)
if role is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
if role != "owner" and not is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the file owner can view shares",
)
shares = db.query(FileShare).filter(FileShare.file_id == file_id).all()
return [_serialize_share(s) for s in shares]
# ---------------------------------------------------------------------------
# Create share
# ---------------------------------------------------------------------------
@router.post("/files/{file_id}/shares", status_code=status.HTTP_201_CREATED)
@require_login
def create_share(
request: Request,
file_id: int,
db: DbSession,
shared_with_user_id: str = Body(..., embed=True),
role: str = Body(FILE_SHARE_ROLE_VIEWER, embed=True),
):
"""Share a document with another user.
Only the file owner may share the document. Sharing with a user
that already has access updates their role instead of creating a
duplicate record.
Path Parameters:
file_id: The ID of the document to share.
Request body (JSON):
shared_with_user_id: The stable user identifier of the recipient.
role: ``"viewer"`` (default) or ``"editor"``.
Returns:
The created or updated share object.
"""
owner_id = get_current_owner_id(request)
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
_require_owner(file_record, owner_id, db)
if role not in FILE_SHARE_ROLES:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"role must be one of: {', '.join(FILE_SHARE_ROLES)}",
)
if not shared_with_user_id or not shared_with_user_id.strip():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="shared_with_user_id must be a non-empty string",
)
shared_with_user_id = shared_with_user_id.strip()
# Cannot share with yourself
if shared_with_user_id == owner_id:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="You cannot share a file with yourself",
)
try:
existing = (
db.query(FileShare)
.filter(FileShare.file_id == file_id, FileShare.shared_with_user_id == shared_with_user_id)
.first()
)
if existing:
# Update role if different
if existing.role != role:
existing.role = role
db.commit()
db.refresh(existing)
logger.info(
"Share updated: file_id=%s, shared_with=%s, role=%s, by owner=%s",
file_id,
shared_with_user_id,
role,
owner_id,
)
return _serialize_share(existing)
share = FileShare(
file_id=file_id,
owner_id=owner_id,
shared_with_user_id=shared_with_user_id,
role=role,
)
db.add(share)
db.commit()
db.refresh(share)
except HTTPException:
raise
except Exception:
db.rollback()
logger.exception("Failed to create share: file_id=%s, shared_with=%s", file_id, shared_with_user_id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to create share",
)
logger.info(
"Share created: id=%s, file_id=%s, shared_with=%s, role=%s, by owner=%s",
share.id,
file_id,
shared_with_user_id,
role,
owner_id,
)
return _serialize_share(share)
# ---------------------------------------------------------------------------
# Update share role
# ---------------------------------------------------------------------------
@router.put("/files/{file_id}/shares/{share_id}")
@require_login
def update_share(
request: Request,
file_id: int,
share_id: int,
db: DbSession,
role: str = Body(..., embed=True),
):
"""Update the role of an existing share.
Only the file owner may change the role of a share.
Path Parameters:
file_id: The ID of the document.
share_id: The ID of the share record to update.
Request body (JSON):
role: New role — ``"viewer"`` or ``"editor"``.
Returns:
The updated share object.
"""
owner_id = get_current_owner_id(request)
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
_require_owner(file_record, owner_id, db)
if role not in FILE_SHARE_ROLES:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"role must be one of: {', '.join(FILE_SHARE_ROLES)}",
)
share = db.query(FileShare).filter(FileShare.id == share_id, FileShare.file_id == file_id).first()
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
try:
share.role = role
db.commit()
db.refresh(share)
except Exception:
db.rollback()
logger.exception("Failed to update share: share_id=%s", share_id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to update share",
)
logger.info("Share updated: id=%s, file_id=%s, new_role=%s, by owner=%s", share_id, file_id, role, owner_id)
return _serialize_share(share)
# ---------------------------------------------------------------------------
# Revoke share
# ---------------------------------------------------------------------------
@router.delete("/files/{file_id}/shares/{share_id}", status_code=status.HTTP_200_OK)
@require_login
def revoke_share(request: Request, file_id: int, share_id: int, db: DbSession):
"""Revoke a share, removing the user's access.
Only the file owner may revoke shares.
Path Parameters:
file_id: The ID of the document.
share_id: The ID of the share record to delete.
Returns:
A success message.
"""
owner_id = get_current_owner_id(request)
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
_require_owner(file_record, owner_id, db)
share = db.query(FileShare).filter(FileShare.id == share_id, FileShare.file_id == file_id).first()
if not share:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
try:
db.delete(share)
db.commit()
except Exception:
db.rollback()
logger.exception("Failed to revoke share: share_id=%s", share_id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to revoke share",
)
logger.info("Share revoked: id=%s, file_id=%s, by owner=%s", share_id, file_id, owner_id)
return {"status": "success", "message": "Share revoked successfully"}
# ---------------------------------------------------------------------------
# List users that the file is already shared with (for the share-picker UI)
# ---------------------------------------------------------------------------
@router.get("/files/{file_id}/shared-with")
@require_login
def list_shared_with(request: Request, file_id: int, db: DbSession):
"""Return the list of users a document is shared with and their roles.
Accessible to any user that has at least viewer access to the file,
so that editors/viewers can see who else has access.
Path Parameters:
file_id: The ID of the document.
Returns:
A list of ``{share_id, user_id, display_name, role}`` objects.
"""
user_id = get_current_owner_id(request)
user = request.session.get("user")
is_admin = isinstance(user, dict) and bool(user.get("is_admin"))
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
if not file_record:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
role = get_file_role(file_record, user_id, db)
if role is None and not is_admin:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
shares = db.query(FileShare).filter(FileShare.file_id == file_id).all()
results = []
for s in shares:
profile = db.query(UserProfile).filter(UserProfile.user_id == s.shared_with_user_id).first()
results.append(
{
"share_id": s.id,
"user_id": s.shared_with_user_id,
"display_name": (profile.display_name if profile and profile.display_name else s.shared_with_user_id),
"role": s.role,
}
)
return results
+49
View File
@@ -1237,3 +1237,52 @@ class DocumentAnnotation(Base):
color = Column(String(20), nullable=True) color = Column(String(20), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now()) created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
# ---------------------------------------------------------------------------
# File sharing
# ---------------------------------------------------------------------------
# Valid roles for FileShare.role
FILE_SHARE_ROLE_VIEWER = "viewer"
FILE_SHARE_ROLE_EDITOR = "editor"
FILE_SHARE_ROLES = (FILE_SHARE_ROLE_VIEWER, FILE_SHARE_ROLE_EDITOR)
class FileShare(Base):
"""Grants a named user access to a ``FileRecord`` owned by someone else.
The ``owner_id`` column records who created the share (must be the file
owner). ``shared_with_user_id`` is the recipient's stable user
identifier (the same kind of string used in ``FileRecord.owner_id``).
Roles
-----
``viewer`` — can read the file, comments, and annotations; may add
comments/annotations; cannot delete or share.
``editor`` — all viewer rights plus the ability to edit document
metadata; cannot delete or re-share.
Only the file owner may create, update, or revoke shares.
"""
__tablename__ = "file_shares"
id = Column(Integer, primary_key=True, index=True)
# The document being shared.
file_id = Column(Integer, ForeignKey(_FILES_ID_FK), nullable=False, index=True)
# The user who granted the share (must match FileRecord.owner_id).
owner_id = Column(String, nullable=False, index=True)
# The user receiving the share.
shared_with_user_id = Column(String, nullable=False, index=True)
# "viewer" or "editor"
role = Column(String(20), nullable=False, default=FILE_SHARE_ROLE_VIEWER)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
__table_args__ = (UniqueConstraint("file_id", "shared_with_user_id", name="uq_file_share_file_user"),)
+91 -5
View File
@@ -11,14 +11,21 @@ import logging
from fastapi import Request from fastapi import Request
from sqlalchemy import or_ from sqlalchemy import or_
from sqlalchemy.orm import Query from sqlalchemy.orm import Query, Session
from sqlalchemy.sql import false from sqlalchemy.sql import false
from app.config import settings from app.config import settings
from app.models import FileRecord from app.models import FILE_SHARE_ROLE_EDITOR, FILE_SHARE_ROLE_VIEWER, FileRecord, FileShare
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Role hierarchy: higher index = more rights
_ROLE_RANK: dict[str, int] = {
FILE_SHARE_ROLE_VIEWER: 1,
FILE_SHARE_ROLE_EDITOR: 2,
"owner": 3,
}
def _owner_id_from_user(user: dict) -> str | None: def _owner_id_from_user(user: dict) -> str | None:
"""Extract the owner identifier from a user dict. """Extract the owner identifier from a user dict.
@@ -93,8 +100,9 @@ def apply_owner_filter(query: Query, request: Request) -> Query:
"""Conditionally filter a ``FileRecord`` query by the current user. """Conditionally filter a ``FileRecord`` query by the current user.
When multi-user mode is enabled, only files whose ``owner_id`` When multi-user mode is enabled, only files whose ``owner_id``
matches the authenticated user are returned. Admin users bypass matches the authenticated user are returned, **plus** any files that
the filter and see all documents. have been explicitly shared with the user via ``FileShare``. Admin
users bypass the filter and see all documents.
When ``unowned_docs_visible_to_all`` is ``True`` (default), documents When ``unowned_docs_visible_to_all`` is ``True`` (default), documents
with ``owner_id IS NULL`` (unclaimed) are also included for every with ``owner_id IS NULL`` (unclaimed) are also included for every
@@ -122,11 +130,89 @@ def apply_owner_filter(query: Query, request: Request) -> Query:
# No authenticated user — return empty result set # No authenticated user — return empty result set
return query.filter(false()) return query.filter(false())
# Build filter: user's own documents # Build filter: user's own documents + documents shared with them
conditions = [FileRecord.owner_id == owner_id] conditions = [FileRecord.owner_id == owner_id]
# Include files explicitly shared with this user
from sqlalchemy import select as sa_select
conditions.append(FileRecord.id.in_(sa_select(FileShare.file_id).where(FileShare.shared_with_user_id == owner_id)))
# Optionally include unclaimed (owner_id IS NULL) documents # Optionally include unclaimed (owner_id IS NULL) documents
if settings.unowned_docs_visible_to_all: if settings.unowned_docs_visible_to_all:
conditions.append(FileRecord.owner_id.is_(None)) conditions.append(FileRecord.owner_id.is_(None))
return query.filter(or_(*conditions)) return query.filter(or_(*conditions))
def get_file_role(file_record: FileRecord, user_id: str | None, db: Session) -> str | None:
"""Return the effective role a user has on a ``FileRecord``.
Roles (in descending order of privilege):
``"owner"`` — the user's ``owner_id`` matches ``file_record.owner_id``,
or multi-user mode is disabled (everyone is effectively an
owner in single-user mode).
``"editor"`` — the user has an explicit ``FileShare`` with role=editor.
``"viewer"`` — the user has an explicit ``FileShare`` with role=viewer,
or the file is unclaimed (``owner_id IS NULL``) and
``unowned_docs_visible_to_all`` is True.
``None`` — no access.
Args:
file_record: The ``FileRecord`` to check.
user_id: The stable identifier of the requesting user.
db: An active SQLAlchemy session.
Returns:
One of ``"owner"``, ``"editor"``, ``"viewer"``, or ``None``.
"""
if not settings.multi_user_enabled:
# Single-user mode: full access for everyone
return "owner"
if user_id is None:
return None
# Owner always has full access
if file_record.owner_id == user_id:
return "owner"
# Unclaimed document — limited access when setting allows it
if file_record.owner_id is None and settings.unowned_docs_visible_to_all:
return FILE_SHARE_ROLE_VIEWER
# Check for an explicit share
share = (
db.query(FileShare)
.filter(FileShare.file_id == file_record.id, FileShare.shared_with_user_id == user_id)
.first()
)
if share:
return share.role
return None
def has_file_role(
file_record: FileRecord,
user_id: str | None,
db: Session,
minimum_role: str = FILE_SHARE_ROLE_VIEWER,
) -> bool:
"""Return ``True`` if the user's effective role meets the minimum required.
Args:
file_record: The document to check.
user_id: Requesting user's stable identifier.
db: Active SQLAlchemy session.
minimum_role: The minimum role required (``"viewer"``, ``"editor"``,
or ``"owner"``).
Returns:
``True`` when the user's role rank is >= the minimum rank.
"""
role = get_file_role(file_record, user_id, db)
if role is None:
return False
return _ROLE_RANK.get(role, 0) >= _ROLE_RANK.get(minimum_role, 0)
+9
View File
@@ -497,6 +497,14 @@ def file_annotations_page(request: Request, file_id: int, db: Session = Depends(
mime = file_record.mime_type or "" mime = file_record.mime_type or ""
is_pdf = mime == "application/pdf" or (file_record.original_filename or "").lower().endswith(".pdf") is_pdf = mime == "application/pdf" or (file_record.original_filename or "").lower().endswith(".pdf")
# Determine the current user's role on this file
from app.utils.user_scope import get_current_owner_id, get_file_role
current_owner_id = get_current_owner_id(request)
user_session = request.session.get("user")
is_admin = isinstance(user_session, dict) and bool(user_session.get("is_admin"))
current_user_role = "owner" if is_admin else (get_file_role(file_record, current_owner_id, db) or "viewer")
return templates.TemplateResponse( return templates.TemplateResponse(
"file_annotations.html", "file_annotations.html",
{ {
@@ -505,6 +513,7 @@ def file_annotations_page(request: Request, file_id: int, db: Session = Depends(
"original_file_exists": original_file_exists, "original_file_exists": original_file_exists,
"processed_file_exists": processed_file_exists, "processed_file_exists": processed_file_exists,
"is_pdf": is_pdf, "is_pdf": is_pdf,
"current_user_role": current_user_role,
}, },
) )
except Exception as e: except Exception as e:
+222
View File
@@ -0,0 +1,222 @@
/**
* sharing.js File sharing management UI.
*
* Renders the current shares for a document and lets the file owner
* add new shares, change roles, or revoke access.
*
* Usage:
* initSharing(fileId, i18n)
*
* The i18n object is expected to contain all keys used below.
*/
/* global fetch */
(function () {
'use strict';
var _fileId = null;
var _i18n = {};
// ── DOM helpers ──────────────────────────────────────────────────────────
function _el(id) {
return document.getElementById(id);
}
function _esc(str) {
return String(str)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#039;');
}
function _t(key) {
return _i18n[key] || key;
}
// ── API helpers ──────────────────────────────────────────────────────────
function _apiUrl(suffix) {
return '/api/files/' + _fileId + suffix;
}
function _fetchShares() {
return fetch(_apiUrl('/shares'), { credentials: 'same-origin' })
.then(function (r) { return r.json(); });
}
function _addShare(userId, role) {
return fetch(_apiUrl('/shares'), {
method: 'POST',
credentials: 'same-origin',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ shared_with_user_id: userId, role: role }),
}).then(function (r) {
return r.json().then(function (body) {
if (!r.ok) throw new Error((body && body.detail) || r.statusText);
return body;
});
});
}
function _updateRole(shareId, role) {
return fetch(_apiUrl('/shares/' + shareId), {
method: 'PUT',
credentials: 'same-origin',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ role: role }),
}).then(function (r) {
return r.json().then(function (body) {
if (!r.ok) throw new Error((body && body.detail) || r.statusText);
return body;
});
});
}
function _revokeShare(shareId) {
return fetch(_apiUrl('/shares/' + shareId), {
method: 'DELETE',
credentials: 'same-origin',
}).then(function (r) {
return r.json().then(function (body) {
if (!r.ok) throw new Error((body && body.detail) || r.statusText);
return body;
});
});
}
// ── Render ───────────────────────────────────────────────────────────────
function _renderShares(shares) {
var list = _el('sharing-list');
if (!list) return;
if (!shares || shares.length === 0) {
list.innerHTML = '<p style="color:#64748b;font-size:0.875rem;">' + _esc(_t('no_shares')) + '</p>';
return;
}
var rows = shares.map(function (s) {
var roleLabel = s.role === 'editor' ? _t('role_editor') : _t('role_viewer');
return (
'<div style="display:flex;align-items:center;justify-content:space-between;gap:0.5rem;padding:0.5rem 0;border-bottom:1px solid #f1f5f9;">' +
'<span style="font-size:0.875rem;color:#334155;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;flex:1;" title="' + _esc(s.user_id) + '">' +
_esc(s.display_name || s.user_id) +
'</span>' +
'<select' +
' data-share-id="' + _esc(s.share_id) + '"' +
' class="sharing-role-select"' +
' aria-label="' + _esc(_t('change_role')) + '"' +
' style="padding:0.25rem 0.5rem;border:1px solid #cbd5e1;border-radius:0.25rem;font-size:0.8rem;background:#fff;"' +
'>' +
'<option value="viewer"' + (s.role === 'viewer' ? ' selected' : '') + '>' + _esc(_t('role_viewer')) + '</option>' +
'<option value="editor"' + (s.role === 'editor' ? ' selected' : '') + '>' + _esc(_t('role_editor')) + '</option>' +
'</select>' +
'<button' +
' data-share-id="' + _esc(s.share_id) + '"' +
' class="sharing-revoke-btn"' +
' aria-label="' + _esc(_t('revoke')) + '"' +
' title="' + _esc(_t('revoke')) + '"' +
' style="padding:0.25rem 0.5rem;background:#fee2e2;color:#b91c1c;border:1px solid #fca5a5;border-radius:0.25rem;font-size:0.8rem;cursor:pointer;"' +
'>' +
'<i class="fas fa-user-minus" aria-hidden="true"></i>' +
'</button>' +
'</div>'
);
});
list.innerHTML = rows.join('');
// Role change handlers
list.querySelectorAll('.sharing-role-select').forEach(function (sel) {
sel.addEventListener('change', function () {
var shareId = sel.getAttribute('data-share-id');
var newRole = sel.value;
_updateRole(shareId, newRole)
.then(function () { _loadAndRender(); })
.catch(function (err) { _showError(err.message); });
});
});
// Revoke handlers
list.querySelectorAll('.sharing-revoke-btn').forEach(function (btn) {
btn.addEventListener('click', function () {
if (!window.confirm(_t('revoke_confirm'))) return;
var shareId = btn.getAttribute('data-share-id');
_revokeShare(shareId)
.then(function () { _loadAndRender(); })
.catch(function (err) { _showError(err.message); });
});
});
}
function _loadAndRender() {
var list = _el('sharing-list');
if (!list) return;
list.innerHTML = '<p style="color:#64748b;font-size:0.875rem;">' + _esc(_t('loading')) + '</p>';
_fetchShares()
.then(function (data) {
// GET /files/{id}/shares returns an array; /files/{id}/shared-with also returns array
var shares = Array.isArray(data) ? data : (data.shares || []);
// Normalise keys: shares list uses share_id, but the shares endpoint returns id
shares = shares.map(function (s) {
return {
share_id: s.share_id || s.id,
user_id: s.user_id || s.shared_with_user_id,
display_name: s.display_name || s.shared_with_user_id || s.user_id,
role: s.role,
};
});
_renderShares(shares);
})
.catch(function (err) {
if (list) list.innerHTML = '<p style="color:#ef4444;font-size:0.875rem;">' + _esc(err.message) + '</p>';
});
}
function _showError(msg) {
var el = _el('sharing-form-error');
if (!el) return;
el.textContent = msg;
el.style.display = 'block';
setTimeout(function () { el.style.display = 'none'; }, 5000);
}
// ── Init ─────────────────────────────────────────────────────────────────
function initSharing(fileId, i18n) {
_fileId = fileId;
_i18n = i18n || {};
_loadAndRender();
var form = _el('sharing-form');
if (!form) return;
form.addEventListener('submit', function (e) {
e.preventDefault();
var userInput = _el('share-user-input');
var roleInput = _el('share-role-input');
var userId = userInput ? userInput.value.trim() : '';
var role = roleInput ? roleInput.value : 'viewer';
if (!userId) {
_showError(_t('error_empty_user'));
return;
}
_addShare(userId, role)
.then(function () {
if (userInput) userInput.value = '';
_loadAndRender();
})
.catch(function (err) { _showError(err.message); });
});
}
// Expose
window.initSharing = initSharing;
})();
+74
View File
@@ -660,6 +660,57 @@
</div> </div>
</div> </div>
<!-- ── Sharing & Permissions ──────────────────────────────────────────── -->
{% if current_user_role == 'owner' %}
<div class="collab-card" id="sharing-panel" style="margin-top:1.5rem;">
<div class="panel-header" style="padding:1rem 1.25rem;border-bottom:1px solid #e2e8f0;display:flex;align-items:center;gap:0.5rem;">
<h3 style="font-size:1rem;font-weight:600;color:#1e293b;margin:0;">
<i class="fas fa-share-alt" aria-hidden="true" style="color:#3b82f6;"></i>
{{ _("sharing.heading") }}
</h3>
</div>
<div style="padding:1rem 1.25rem;">
<!-- Current shares list -->
<div id="sharing-list" aria-live="polite" style="margin-bottom:1rem;">
<p style="color:#64748b;font-size:0.875rem;">{{ _("sharing.loading") }}</p>
</div>
<!-- Add share form -->
<form id="sharing-form" aria-label="{{ _('sharing.add_share') }}" style="display:flex;gap:0.5rem;flex-wrap:wrap;align-items:flex-end;">
<div style="flex:1;min-width:180px;">
<label for="share-user-input" style="display:block;font-size:0.8rem;color:#475569;margin-bottom:0.25rem;">
{{ _("sharing.user_id_label") }}
</label>
<input
type="text"
id="share-user-input"
placeholder="{{ _('sharing.user_id_placeholder') }}"
style="width:100%;padding:0.5rem 0.75rem;border:1px solid #cbd5e1;border-radius:0.375rem;font-size:0.875rem;"
aria-label="{{ _('sharing.user_id_label') }}"
/>
</div>
<div style="min-width:120px;">
<label for="share-role-input" style="display:block;font-size:0.8rem;color:#475569;margin-bottom:0.25rem;">
{{ _("sharing.role_label") }}
</label>
<select
id="share-role-input"
style="width:100%;padding:0.5rem 0.75rem;border:1px solid #cbd5e1;border-radius:0.375rem;font-size:0.875rem;background:#fff;"
aria-label="{{ _('sharing.role_label') }}"
>
<option value="viewer">{{ _("sharing.role_viewer") }}</option>
<option value="editor">{{ _("sharing.role_editor") }}</option>
</select>
</div>
<button type="submit" style="padding:0.5rem 1rem;background:#3b82f6;color:#fff;border:none;border-radius:0.375rem;font-size:0.875rem;font-weight:600;cursor:pointer;min-height:2.25rem;" aria-label="{{ _('sharing.add_share') }}">
<i class="fas fa-user-plus" aria-hidden="true"></i> {{ _("sharing.add_share") }}
</button>
</form>
<p id="sharing-form-error" style="color:#ef4444;font-size:0.8rem;margin-top:0.5rem;display:none;" role="alert"></p>
</div>
</div>
{% endif %}
{% else %} {% else %}
<!-- file is None without error --> <!-- file is None without error -->
<div class="error-box">Document not found.</div> <div class="error-box">Document not found.</div>
@@ -713,6 +764,25 @@
if (typeof initAnnotations === 'function') { if (typeof initAnnotations === 'function') {
initAnnotations(fileId, userId, annotationsI18n); initAnnotations(fileId, userId, annotationsI18n);
} }
{% if current_user_role == 'owner' %}
if (typeof initSharing === 'function') {
initSharing(fileId, {
heading: {{ _("sharing.heading") | tojson }},
loading: {{ _("sharing.loading") | tojson }},
no_shares: {{ _("sharing.no_shares") | tojson }},
add_share: {{ _("sharing.add_share") | tojson }},
role_viewer: {{ _("sharing.role_viewer") | tojson }},
role_editor: {{ _("sharing.role_editor") | tojson }},
revoke: {{ _("sharing.revoke") | tojson }},
revoke_confirm: {{ _("sharing.revoke_confirm") | tojson }},
user_id_label: {{ _("sharing.user_id_label") | tojson }},
user_id_placeholder: {{ _("sharing.user_id_placeholder") | tojson }},
role_label: {{ _("sharing.role_label") | tojson }},
error_empty_user: {{ _("sharing.error_empty_user") | tojson }},
change_role: {{ _("sharing.change_role") | tojson }},
});
}
{% endif %}
}) })
.catch(function () { .catch(function () {
// Auth disabled — initialise with empty user // Auth disabled — initialise with empty user
@@ -722,6 +792,10 @@
{% endif %} {% endif %}
}); });
</script> </script>
<!-- Sharing JS -->
{% if current_user_role == 'owner' %}
<script src="{{ url_for('static', path='js/sharing.js') }}" defer></script>
{% endif %}
<!-- ── EmbedPDF Viewer init ── --> <!-- ── EmbedPDF Viewer init ── -->
{% if file and is_pdf and (processed_file_exists or original_file_exists) %} {% if file and is_pdf and (processed_file_exists or original_file_exists) %}
<script async type="module"> <script async type="module">
+13
View File
@@ -1743,6 +1743,19 @@
"shared.table_aria": "Shared links", "shared.table_aria": "Shared links",
"shared.unlimited_placeholder": "Unlimited", "shared.unlimited_placeholder": "Unlimited",
"shared.your_links": "Your Shared Links", "shared.your_links": "Your Shared Links",
"sharing.add_share": "Share",
"sharing.change_role": "Change role",
"sharing.error_empty_user": "Please enter a user ID to share with.",
"sharing.heading": "Share with Users",
"sharing.loading": "Loading shares…",
"sharing.no_shares": "Not shared with anyone yet.",
"sharing.revoke": "Revoke access",
"sharing.revoke_confirm": "Remove this user's access to the file?",
"sharing.role_editor": "Editor",
"sharing.role_label": "Role",
"sharing.role_viewer": "Viewer",
"sharing.user_id_label": "User ID or email",
"sharing.user_id_placeholder": "e.g. alice@example.com",
"similarity.backfill_auto": "The background task will compute them automatically every 5 minutes, or you can", "similarity.backfill_auto": "The background task will compute them automatically every 5 minutes, or you can",
"similarity.files_missing_text": "file(s) have OCR text but no embedding yet.", "similarity.files_missing_text": "file(s) have OCR text but no embedding yet.",
"similarity.find_pairs_btn": "Find Pairs", "similarity.find_pairs_btn": "Find Pairs",
@@ -0,0 +1,56 @@
"""Add file_shares table for per-user document sharing and role-based access.
Revision ID: 042_add_file_shares
Revises: 041_add_document_comments_and_annotations
Create Date: 2026-03-22
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "042_add_file_shares"
down_revision: str = "041_add_document_comments_and_annotations"
branch_labels = None
depends_on = None
def upgrade() -> None:
"""Create file_shares table."""
conn = op.get_bind()
inspector = sa.inspect(conn)
existing_tables = set(inspector.get_table_names())
if "file_shares" not in existing_tables:
op.create_table(
"file_shares",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("file_id", sa.Integer(), nullable=False),
sa.Column("owner_id", sa.String(), nullable=False),
sa.Column("shared_with_user_id", sa.String(), nullable=False),
sa.Column("role", sa.String(20), nullable=False, server_default="viewer"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
onupdate=sa.func.now(),
),
sa.ForeignKeyConstraint(["file_id"], ["files.id"]),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("file_id", "shared_with_user_id", name="uq_file_share_file_user"),
)
op.create_index("ix_file_shares_id", "file_shares", ["id"])
op.create_index("ix_file_shares_file_id", "file_shares", ["file_id"])
op.create_index("ix_file_shares_owner_id", "file_shares", ["owner_id"])
op.create_index("ix_file_shares_shared_with_user_id", "file_shares", ["shared_with_user_id"])
def downgrade() -> None:
"""Drop file_shares table."""
op.drop_index("ix_file_shares_shared_with_user_id", table_name="file_shares")
op.drop_index("ix_file_shares_owner_id", table_name="file_shares")
op.drop_index("ix_file_shares_file_id", table_name="file_shares")
op.drop_index("ix_file_shares_id", table_name="file_shares")
op.drop_table("file_shares")
+1
View File
@@ -69,6 +69,7 @@ from app.models import ( # noqa: F401, E402
DocumentComment, DocumentComment,
DocumentMetadata, DocumentMetadata,
FileRecord, FileRecord,
FileShare,
Pipeline, Pipeline,
PipelineRoutingRule, PipelineRoutingRule,
PipelineStep, PipelineStep,
+691
View File
@@ -0,0 +1,691 @@
"""Tests for the file sharing API (FileShare model and /api/files/{id}/shares endpoints)."""
import pytest
from app.models import FILE_SHARE_ROLE_EDITOR, FILE_SHARE_ROLE_VIEWER, FileRecord, FileShare, UserProfile
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _create_file(db_session, owner_id="owner1") -> FileRecord:
"""Create a minimal owned FileRecord."""
f = FileRecord(
owner_id=owner_id,
filehash="sharehash",
original_filename="shared.pdf",
local_filename="shared.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(f)
db_session.commit()
db_session.refresh(f)
return f
def _create_unowned_file(db_session) -> FileRecord:
"""Create a FileRecord with no owner."""
f = FileRecord(
owner_id=None,
filehash="unownedhash",
original_filename="unowned.pdf",
local_filename="unowned.pdf",
file_size=512,
mime_type="application/pdf",
)
db_session.add(f)
db_session.commit()
db_session.refresh(f)
return f
def _create_profile(db_session, user_id: str, display_name: str | None = None) -> UserProfile:
p = UserProfile(user_id=user_id, display_name=display_name)
db_session.add(p)
db_session.commit()
db_session.refresh(p)
return p
# ---------------------------------------------------------------------------
# get_file_role helper
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestGetFileRole:
"""Tests for the get_file_role() utility."""
def test_owner_returns_owner(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import get_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert get_file_role(f, "alice", db_session) == "owner"
def test_non_owner_no_share_returns_none(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import get_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert get_file_role(f, "bob", db_session) is None
def test_shared_viewer_returns_viewer(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import get_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role=FILE_SHARE_ROLE_VIEWER)
db_session.add(share)
db_session.commit()
assert get_file_role(f, "bob", db_session) == FILE_SHARE_ROLE_VIEWER
def test_shared_editor_returns_editor(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import get_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="carol", role=FILE_SHARE_ROLE_EDITOR)
db_session.add(share)
db_session.commit()
assert get_file_role(f, "carol", db_session) == FILE_SHARE_ROLE_EDITOR
def test_unowned_file_returns_viewer_when_setting_allows(self, db_session, monkeypatch):
from app.utils import user_scope
monkeypatch.setattr(user_scope.settings, "multi_user_enabled", True)
monkeypatch.setattr(user_scope.settings, "unowned_docs_visible_to_all", True)
f = _create_unowned_file(db_session)
role = user_scope.get_file_role(f, "anyone", db_session)
assert role == FILE_SHARE_ROLE_VIEWER
def test_none_user_returns_none(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import get_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert get_file_role(f, None, db_session) is None
# ---------------------------------------------------------------------------
# has_file_role helper
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestHasFileRole:
"""Tests for the has_file_role() utility."""
def test_owner_satisfies_viewer(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert has_file_role(f, "alice", db_session, minimum_role="viewer") is True
def test_owner_satisfies_editor(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert has_file_role(f, "alice", db_session, minimum_role="editor") is True
def test_owner_satisfies_owner(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert has_file_role(f, "alice", db_session, minimum_role="owner") is True
def test_viewer_does_not_satisfy_editor(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role=FILE_SHARE_ROLE_VIEWER)
db_session.add(share)
db_session.commit()
assert has_file_role(f, "bob", db_session, minimum_role="editor") is False
def test_editor_satisfies_viewer(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="carol", role=FILE_SHARE_ROLE_EDITOR)
db_session.add(share)
db_session.commit()
assert has_file_role(f, "carol", db_session, minimum_role="viewer") is True
def test_no_access_returns_false(self, db_session, monkeypatch):
from app.config import settings as real_settings
from app.utils.user_scope import has_file_role
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
assert has_file_role(f, "stranger", db_session) is False
# ---------------------------------------------------------------------------
# List shares
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestListShares:
"""Tests for GET /api/files/{file_id}/shares."""
def test_owner_can_list_shares(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role=FILE_SHARE_ROLE_VIEWER)
db_session.add(share)
db_session.commit()
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
resp = client.get(f"/api/files/{f.id}/shares")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, list)
assert len(data) == 1
assert data[0]["shared_with_user_id"] == "bob"
assert data[0]["role"] == FILE_SHARE_ROLE_VIEWER
def test_non_owner_cannot_list_shares(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
f = _create_file(db_session, owner_id="alice")
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "bob")
# bob has no access to alice's file — the get_file_role call in list_shares
# will return None for bob, giving 404 not 403 (file not found for bob)
resp = client.get(f"/api/files/{f.id}/shares")
assert resp.status_code in (403, 404)
def test_list_shares_file_not_found(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
resp = client.get("/api/files/99999/shares")
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# Create share
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestCreateShare:
"""Tests for POST /api/files/{file_id}/shares."""
def test_owner_can_share_with_viewer(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "bob", "role": "viewer"},
)
assert resp.status_code == 201
data = resp.json()
assert data["shared_with_user_id"] == "bob"
assert data["role"] == "viewer"
assert data["file_id"] == f.id
def test_owner_can_share_with_editor(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "carol", "role": "editor"},
)
assert resp.status_code == 201
assert resp.json()["role"] == "editor"
def test_non_owner_cannot_share(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "bob")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "carol", "role": "viewer"},
)
# bob doesn't own the file; get_file_role returns None → 404 for non-owner
assert resp.status_code in (403, 404)
def test_share_with_self_rejected(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "alice", "role": "viewer"},
)
assert resp.status_code == 422
def test_invalid_role_rejected(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "bob", "role": "admin"},
)
assert resp.status_code == 422
def test_empty_user_id_rejected(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": " ", "role": "viewer"},
)
assert resp.status_code == 422
def test_duplicate_share_updates_role(self, client, db_session, monkeypatch):
"""Creating a share for an already-shared user updates the role."""
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
resp = client.post(
f"/api/files/{f.id}/shares",
json={"shared_with_user_id": "bob", "role": "editor"},
)
assert resp.status_code == 201
assert resp.json()["role"] == "editor"
# ---------------------------------------------------------------------------
# Update share role
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestUpdateShare:
"""Tests for PUT /api/files/{file_id}/shares/{share_id}."""
def test_owner_can_update_role(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
db_session.refresh(share)
resp = client.put(
f"/api/files/{f.id}/shares/{share.id}",
json={"role": "editor"},
)
assert resp.status_code == 200
assert resp.json()["role"] == "editor"
def test_non_owner_cannot_update_role(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "bob")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
db_session.refresh(share)
resp = client.put(
f"/api/files/{f.id}/shares/{share.id}",
json={"role": "editor"},
)
assert resp.status_code in (403, 404)
def test_invalid_role_rejected(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
db_session.refresh(share)
resp = client.put(
f"/api/files/{f.id}/shares/{share.id}",
json={"role": "superuser"},
)
assert resp.status_code == 422
def test_share_not_found(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.put(
f"/api/files/{f.id}/shares/99999",
json={"role": "editor"},
)
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# Revoke share
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestRevokeShare:
"""Tests for DELETE /api/files/{file_id}/shares/{share_id}."""
def test_owner_can_revoke(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
db_session.refresh(share)
resp = client.delete(f"/api/files/{f.id}/shares/{share.id}")
assert resp.status_code == 200
assert resp.json()["status"] == "success"
# Confirm the share is gone
db_session.expire_all()
assert db_session.query(FileShare).filter(FileShare.id == share.id).first() is None
def test_non_owner_cannot_revoke(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "carol")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
db_session.refresh(share)
resp = client.delete(f"/api/files/{f.id}/shares/{share.id}")
assert resp.status_code in (403, 404)
def test_revoke_not_found(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.delete(f"/api/files/{f.id}/shares/99999")
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# List shared-with
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestListSharedWith:
"""Tests for GET /api/files/{file_id}/shared-with."""
def test_owner_can_see_shared_with(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
_create_profile(db_session, "bob", "Bob Smith")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
resp = client.get(f"/api/files/{f.id}/shared-with")
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["user_id"] == "bob"
assert data[0]["display_name"] == "Bob Smith"
assert data[0]["role"] == "viewer"
def test_viewer_can_see_shared_with(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "bob")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
resp = client.get(f"/api/files/{f.id}/shared-with")
assert resp.status_code == 200
def test_unauthorized_user_gets_404(self, client, db_session, monkeypatch):
import app.api.sharing as sharing_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(sharing_mod, "get_current_owner_id", lambda req: "stranger")
f = _create_file(db_session, owner_id="alice")
resp = client.get(f"/api/files/{f.id}/shared-with")
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# Auto-share on mention
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestAutoShareOnMention:
"""Tests that @mentioning a user in a comment auto-shares the file."""
def test_mention_auto_shares_with_viewer(self, client, db_session, monkeypatch):
"""When multi_user_enabled is True, mentioning a user auto-shares the file."""
import app.api.comments as comments_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(comments_mod, "get_current_owner_id", lambda req: "alice")
monkeypatch.setattr(comments_mod, "get_current_user_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/comments",
json={"body": "Hey @bob, please look at this."},
)
assert resp.status_code == 201
# bob should now have a viewer share on the file
share = (
db_session.query(FileShare)
.filter(FileShare.file_id == f.id, FileShare.shared_with_user_id == "bob")
.first()
)
assert share is not None
assert share.role == FILE_SHARE_ROLE_VIEWER
def test_mention_does_not_duplicate_share(self, client, db_session, monkeypatch):
"""Mentioning a user that already has a share does not create a duplicate."""
import app.api.comments as comments_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(comments_mod, "get_current_owner_id", lambda req: "alice")
monkeypatch.setattr(comments_mod, "get_current_user_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
existing = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="editor")
db_session.add(existing)
db_session.commit()
existing_id = existing.id
resp = client.post(
f"/api/files/{f.id}/comments",
json={"body": "Hey @bob again!"},
)
assert resp.status_code == 201
shares = (
db_session.query(FileShare).filter(FileShare.file_id == f.id, FileShare.shared_with_user_id == "bob").all()
)
assert len(shares) == 1
assert shares[0].id == existing_id
assert shares[0].role == "editor" # role unchanged
def test_mention_skipped_when_single_user_mode(self, client, db_session, monkeypatch):
import app.api.comments as comments_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", False)
monkeypatch.setattr(comments_mod, "get_current_owner_id", lambda req: "alice")
monkeypatch.setattr(comments_mod, "get_current_user_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.post(
f"/api/files/{f.id}/comments",
json={"body": "Hey @carol, look here."},
)
assert resp.status_code == 201
share = (
db_session.query(FileShare)
.filter(FileShare.file_id == f.id, FileShare.shared_with_user_id == "carol")
.first()
)
assert share is None
# ---------------------------------------------------------------------------
# Delete file owner-only enforcement
# ---------------------------------------------------------------------------
@pytest.mark.unit
class TestDeleteFileOwnerOnly:
"""Ensure non-owners (shared viewers/editors) cannot delete files."""
def test_owner_can_delete_in_multi_user_mode(self, client, db_session, monkeypatch):
import app.api.files as files_mod
import app.utils.user_scope as user_scope_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(real_settings, "allow_file_delete", True)
monkeypatch.setattr(files_mod, "get_current_owner_id", lambda req: "alice")
monkeypatch.setattr(user_scope_mod, "get_current_owner_id", lambda req: "alice")
f = _create_file(db_session, owner_id="alice")
resp = client.delete(f"/api/files/{f.id}")
assert resp.status_code == 200
def test_viewer_cannot_delete(self, client, db_session, monkeypatch):
import app.api.files as files_mod
import app.utils.user_scope as user_scope_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(real_settings, "allow_file_delete", True)
monkeypatch.setattr(files_mod, "get_current_owner_id", lambda req: "bob")
monkeypatch.setattr(user_scope_mod, "get_current_owner_id", lambda req: "bob")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="bob", role="viewer")
db_session.add(share)
db_session.commit()
resp = client.delete(f"/api/files/{f.id}")
assert resp.status_code == 403
def test_editor_cannot_delete(self, client, db_session, monkeypatch):
import app.api.files as files_mod
import app.utils.user_scope as user_scope_mod
from app.config import settings as real_settings
monkeypatch.setattr(real_settings, "multi_user_enabled", True)
monkeypatch.setattr(real_settings, "allow_file_delete", True)
monkeypatch.setattr(files_mod, "get_current_owner_id", lambda req: "carol")
monkeypatch.setattr(user_scope_mod, "get_current_owner_id", lambda req: "carol")
f = _create_file(db_session, owner_id="alice")
share = FileShare(file_id=f.id, owner_id="alice", shared_with_user_id="carol", role="editor")
db_session.add(share)
db_session.commit()
resp = client.delete(f"/api/files/{f.id}")
assert resp.status_code == 403