refac

backend/open_webui/models/access_grants.py

@@ -11,6 +11,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 log = logging.getLogger(__name__)
 
+PRINCIPAL_TYPE_ANYONE = 'anyone'
+PRINCIPAL_TYPE_GROUP = 'group'
+PRINCIPAL_TYPE_USER = 'user'
+WILDCARD_PRINCIPAL_ID = '*'
+
 
 ####################
 # AccessGrant DB Schema
@@ -23,7 +28,7 @@ class AccessGrant(Base):
     id = Column(Text, primary_key=True)
     resource_type = Column(Text, nullable=False)  # "knowledge", "model", "prompt", "tool", "note", "channel", "file"
     resource_id = Column(Text, nullable=False)
-    principal_type = Column(Text, nullable=False)  # "user" or "group"
+    principal_type = Column(Text, nullable=False)  # "user", "group", or "anyone"
     principal_id = Column(Text, nullable=False)  # user_id, group_id, or "*" (wildcard for public)
     permission = Column(Text, nullable=False)  # "read" or "write"
     created_at = Column(BigInteger, nullable=False)
@@ -163,12 +168,16 @@ def normalize_access_grants(access_grants: Optional[list]) -> list[dict]:
         principal_id = grant.get('principal_id')
         permission = grant.get('permission')
 
-        if principal_type not in ('user', 'group'):
+        if principal_type not in (PRINCIPAL_TYPE_USER, PRINCIPAL_TYPE_GROUP, PRINCIPAL_TYPE_ANYONE):
             continue
         if permission not in ('read', 'write'):
             continue
         if not isinstance(principal_id, str) or not principal_id:
             continue
+        if principal_type == PRINCIPAL_TYPE_ANYONE and (
+            principal_id != WILDCARD_PRINCIPAL_ID or permission != 'read'
+        ):
+            continue
 
         key = (principal_type, principal_id, permission)
         deduped[key] = {
@@ -186,7 +195,11 @@ def has_public_read_access_grant(access_grants: Optional[list]) -> bool:
     Returns True when a direct grant list includes wildcard public-read.
     """
     for grant in normalize_access_grants(access_grants):
-        if grant['principal_type'] == 'user' and grant['principal_id'] == '*' and grant['permission'] == 'read':
+        if (
+            grant['principal_type'] == PRINCIPAL_TYPE_USER
+            and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
+            and grant['permission'] == 'read'
+        ):
             return True
     return False
 
@@ -196,7 +209,25 @@ def has_public_write_access_grant(access_grants: Optional[list]) -> bool:
     Returns True when a direct grant list includes wildcard public-write.
     """
     for grant in normalize_access_grants(access_grants):
-        if grant['principal_type'] == 'user' and grant['principal_id'] == '*' and grant['permission'] == 'write':
+        if (
+            grant['principal_type'] == PRINCIPAL_TYPE_USER
+            and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
+            and grant['permission'] == 'write'
+        ):
+            return True
+    return False
+
+
+def has_anyone_read_access_grant(access_grants: Optional[list]) -> bool:
+    """
+    Returns True when a direct grant list includes no-auth anyone-read.
+    """
+    for grant in normalize_access_grants(access_grants):
+        if (
+            grant['principal_type'] == PRINCIPAL_TYPE_ANYONE
+            and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
+            and grant['permission'] == 'read'
+        ):
             return True
     return False
 
@@ -206,7 +237,7 @@ def has_user_access_grant(access_grants: Optional[list]) -> bool:
     Returns True when a direct grant list includes any non-wildcard user grant.
     """
     for grant in normalize_access_grants(access_grants):
-        if grant['principal_type'] == 'user' and grant['principal_id'] != '*':
+        if grant['principal_type'] == PRINCIPAL_TYPE_USER and grant['principal_id'] != WILDCARD_PRINCIPAL_ID:
             return True
     return False
 
@@ -223,12 +254,27 @@ def strip_user_access_grants(access_grants: Optional[list]) -> list:
         for grant in access_grants
         if not (
             (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
-            == 'user'
-            and (grant.get('principal_id') if isinstance(grant, dict) else getattr(grant, 'principal_id', None)) != '*'
+            == PRINCIPAL_TYPE_USER
+            and (grant.get('principal_id') if isinstance(grant, dict) else getattr(grant, 'principal_id', None))
+            != WILDCARD_PRINCIPAL_ID
         )
     ]
 
 
+def strip_anyone_access_grants(access_grants: Optional[list]) -> list:
+    """
+    Remove no-auth anyone grants from the list.
+    """
+    if not access_grants:
+        return []
+    return [
+        grant
+        for grant in access_grants
+        if (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
+        != PRINCIPAL_TYPE_ANYONE
+    ]
+
+
 def grants_to_access_control(grants: list) -> Optional[dict]:
     """
     Convert a list of grant objects (AccessGrantModel or AccessGrantResponse)
@@ -493,6 +539,28 @@ class AccessGrantsTable:
                 result_dict[g.resource_id].append(AccessGrantModel.model_validate(g))
             return result_dict
 
+    async def has_anyone_access(
+        self,
+        resource_type: str,
+        resource_id: str,
+        permission: str = 'read',
+        db: Optional[AsyncSession] = None,
+    ) -> bool:
+        """Check for a no-auth anyone:* grant. Callers must opt in explicitly."""
+        async with get_async_db_context(db) as db:
+            result = await db.execute(
+                select(AccessGrant)
+                .filter(
+                    AccessGrant.resource_type == resource_type,
+                    AccessGrant.resource_id == resource_id,
+                    AccessGrant.principal_type == PRINCIPAL_TYPE_ANYONE,
+                    AccessGrant.principal_id == WILDCARD_PRINCIPAL_ID,
+                    AccessGrant.permission == permission,
+                )
+                .limit(1)
+            )
+            return result.scalars().first() is not None
+
     async def has_access(
         self,
         user_id: str,