server: add privilege posts:list:unsafe, filter unsafe posts for users without privilege

server/config.yaml.dist

@@ -97,6 +97,7 @@ privileges:
     'posts:create:anonymous':       regular
     'posts:create:identified':      regular
     'posts:list':                   anonymous
+    'posts:list:unsafe':            regular
     'posts:reverse_search':         regular
     'posts:view':                   anonymous
     'posts:view:featured':          anonymous

server/szurubooru/search/configs/post_search_config.py

@@ -3,7 +3,7 @@ from typing import Any, Dict, Optional, Tuple
 import sqlalchemy as sa
 
 from szurubooru import db, errors, model
-from szurubooru.func import util
+from szurubooru.func import auth, util
 from szurubooru.search import criteria, tokens
 from szurubooru.search.configs import util as search_util
 from szurubooru.search.configs.base_search_config import (
@@ -150,6 +150,15 @@ def _category_filter(
     return query.filter(expr)
 
 
+def _safety_filter(
+    query: SaQuery, criterion: Optional[criteria.BaseCriterion], negated: bool
+) -> SaQuery:
+    assert criterion
+    return search_util.create_str_filter(
+        model.Post.safety, _safety_transformer
+    )(query, criterion, negated)
+
+
 class PostSearchConfig(BaseSearchConfig):
     def __init__(self) -> None:
         self.user = None  # type: Optional[model.User]
@@ -208,6 +217,15 @@ class PostSearchConfig(BaseSearchConfig):
         return db.session.query(model.Post)
 
     def finalize_query(self, query: SaQuery) -> SaQuery:
+        if self.user and not auth.has_privilege(self.user, "posts:list:unsafe"):
+            # exclude unsafe posts:
+            query = _safety_filter(
+                query,
+                criteria.PlainCriterion(
+                    model.Post.SAFETY_UNSAFE, model.Post.SAFETY_UNSAFE
+                ),
+                negated=True,
+            )
         return query.order_by(model.Post.post_id.desc())
 
     @property
@@ -363,12 +381,7 @@ class PostSearchConfig(BaseSearchConfig):
                         model.Post.last_feature_time
                     ),
                 ),
-                (
-                    ["safety", "rating"],
-                    search_util.create_str_filter(
-                        model.Post.safety, _safety_transformer
-                    ),
-                ),
+                (["safety", "rating"], _safety_filter),
                 (["note-text"], _note_filter),
                 (
                     ["flag"],

server/szurubooru/tests/search/configs/test_post_search_config.py

@@ -3,6 +3,12 @@ from datetime import datetime
 import pytest
 
 from szurubooru import db, errors, model, search
+from szurubooru.func import cache
+
+
+@pytest.fixture(autouse=True)
+def purge_cache():
+    cache.purge()
 
 
 @pytest.fixture
@@ -915,3 +921,36 @@ def test_search_by_tag_category(
     )
     db.session.flush()
     verify_unpaged(input, expected_post_ids)
+
+
+def test_filter_unsafe_without_privilege(
+    auth_executor,
+    verify_unpaged,
+    post_factory,
+    config_injector,
+):
+    config_injector(
+        {
+            "privileges": {
+                "posts:list:unsafe": model.User.RANK_REGULAR,
+            }
+        }
+    )
+    post1 = post_factory(id=1)
+    post2 = post_factory(id=2, safety=model.Post.SAFETY_SKETCHY)
+    post3 = post_factory(id=3, safety=model.Post.SAFETY_UNSAFE)
+    db.session.add_all([post1, post2, post3])
+    db.session.flush()
+    user = auth_executor()
+    user.rank = model.User.RANK_ANONYMOUS
+    verify_unpaged("", [1, 2])
+    verify_unpaged("safety:safe", [1])
+    verify_unpaged("safety:safe,sketchy", [1, 2])
+    verify_unpaged("safety:safe,sketchy,unsafe", [1, 2])
+    # adjust user's rank and retry
+    user.rank = model.User.RANK_REGULAR
+    cache.purge()
+    verify_unpaged("", [1, 2, 3])
+    verify_unpaged("safety:safe", [1])
+    verify_unpaged("safety:safe,sketchy", [1, 2])
+    verify_unpaged("safety:safe,sketchy,unsafe", [1, 2, 3])