enh: read only notes
This commit is contained in:
@@ -96,11 +96,86 @@ class NoteTable:
|
||||
def _has_permission(self, db, query, filter: dict, permission: str = "read"):
|
||||
group_ids = filter.get("group_ids", [])
|
||||
user_id = filter.get("user_id")
|
||||
|
||||
dialect_name = db.bind.dialect.name
|
||||
|
||||
# Public access
|
||||
conditions = []
|
||||
|
||||
# Handle read_only permission separately
|
||||
if permission == "read_only":
|
||||
# For read_only, we want items where:
|
||||
# 1. User has explicit read permission (via groups or user-level)
|
||||
# 2. BUT does NOT have write permission
|
||||
# 3. Public items are NOT considered read_only
|
||||
|
||||
read_conditions = []
|
||||
|
||||
# Group-level read permission
|
||||
if group_ids:
|
||||
group_read_conditions = []
|
||||
for gid in group_ids:
|
||||
if dialect_name == "sqlite":
|
||||
group_read_conditions.append(
|
||||
Note.access_control["read"]["group_ids"].contains([gid])
|
||||
)
|
||||
elif dialect_name == "postgresql":
|
||||
group_read_conditions.append(
|
||||
cast(
|
||||
Note.access_control["read"]["group_ids"],
|
||||
JSONB,
|
||||
).contains([gid])
|
||||
)
|
||||
|
||||
if group_read_conditions:
|
||||
read_conditions.append(or_(*group_read_conditions))
|
||||
|
||||
# Combine read conditions
|
||||
if read_conditions:
|
||||
has_read = or_(*read_conditions)
|
||||
else:
|
||||
# If no read conditions, return empty result
|
||||
return query.filter(False)
|
||||
|
||||
# Now exclude items where user has write permission
|
||||
write_exclusions = []
|
||||
|
||||
# Exclude items owned by user (they have implicit write)
|
||||
if user_id:
|
||||
write_exclusions.append(Note.user_id != user_id)
|
||||
|
||||
# Exclude items where user has explicit write permission via groups
|
||||
if group_ids:
|
||||
group_write_conditions = []
|
||||
for gid in group_ids:
|
||||
if dialect_name == "sqlite":
|
||||
group_write_conditions.append(
|
||||
Note.access_control["write"]["group_ids"].contains([gid])
|
||||
)
|
||||
elif dialect_name == "postgresql":
|
||||
group_write_conditions.append(
|
||||
cast(
|
||||
Note.access_control["write"]["group_ids"],
|
||||
JSONB,
|
||||
).contains([gid])
|
||||
)
|
||||
|
||||
if group_write_conditions:
|
||||
# User should NOT have write permission
|
||||
write_exclusions.append(~or_(*group_write_conditions))
|
||||
|
||||
# Exclude public items (items without access_control)
|
||||
write_exclusions.append(Note.access_control.isnot(None))
|
||||
write_exclusions.append(cast(Note.access_control, String) != "null")
|
||||
|
||||
# Combine: has read AND does not have write AND not public
|
||||
if write_exclusions:
|
||||
query = query.filter(and_(has_read, *write_exclusions))
|
||||
else:
|
||||
query = query.filter(has_read)
|
||||
|
||||
return query
|
||||
|
||||
# Original logic for other permissions (read, write, etc.)
|
||||
# Public access conditions
|
||||
if group_ids or user_id:
|
||||
conditions.extend(
|
||||
[
|
||||
@@ -109,7 +184,7 @@ class NoteTable:
|
||||
]
|
||||
)
|
||||
|
||||
# User-level permission
|
||||
# User-level permission (owner has all permissions)
|
||||
if user_id:
|
||||
conditions.append(Note.user_id == user_id)
|
||||
|
||||
@@ -191,11 +266,16 @@ class NoteTable:
|
||||
query = query.filter(Note.user_id != user_id)
|
||||
|
||||
# Apply access control filtering
|
||||
if "permission" in filter:
|
||||
permission = filter["permission"]
|
||||
else:
|
||||
permission = "write"
|
||||
|
||||
query = self._has_permission(
|
||||
db,
|
||||
query,
|
||||
filter,
|
||||
permission="write",
|
||||
permission=permission,
|
||||
)
|
||||
|
||||
order_by = filter.get("order_by")
|
||||
|
||||
@@ -84,6 +84,7 @@ async def search_notes(
|
||||
request: Request,
|
||||
query: Optional[str] = None,
|
||||
view_option: Optional[str] = None,
|
||||
permission: Optional[str] = None,
|
||||
order_by: Optional[str] = None,
|
||||
direction: Optional[str] = None,
|
||||
page: Optional[int] = 1,
|
||||
@@ -108,6 +109,8 @@ async def search_notes(
|
||||
filter["query"] = query
|
||||
if view_option:
|
||||
filter["view_option"] = view_option
|
||||
if permission:
|
||||
filter["permission"] = permission
|
||||
if order_by:
|
||||
filter["order_by"] = order_by
|
||||
if direction:
|
||||
@@ -156,7 +159,11 @@ async def create_new_note(
|
||||
############################
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=Optional[NoteModel])
|
||||
class NoteResponse(NoteModel):
|
||||
write_access: bool = False
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=Optional[NoteResponse])
|
||||
async def get_note_by_id(request: Request, id: str, user=Depends(get_verified_user)):
|
||||
if user.role != "admin" and not has_permission(
|
||||
user.id, "features.notes", request.app.state.config.USER_PERMISSIONS
|
||||
@@ -180,7 +187,11 @@ async def get_note_by_id(request: Request, id: str, user=Depends(get_verified_us
|
||||
status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT()
|
||||
)
|
||||
|
||||
return note
|
||||
write_access = has_access(
|
||||
user.id, type="write", access_control=note.access_control, strict=False
|
||||
)
|
||||
|
||||
return NoteResponse(**note.model_dump(), write_access=write_access)
|
||||
|
||||
|
||||
############################
|
||||
|
||||
Reference in New Issue
Block a user