enh: access grant level perms

This commit is contained in:
Timothy Jaeryang Baek
2026-02-23 15:49:05 -06:00
parent a52e6c2d57
commit 3d99de6771
22 changed files with 201 additions and 8 deletions
+7
View File
@@ -1433,6 +1433,10 @@ USER_PERMISSIONS_NOTES_ALLOW_PUBLIC_SHARING = (
== "true"
)
USER_PERMISSIONS_ACCESS_GRANTS_ALLOW_USERS = (
os.environ.get("USER_PERMISSIONS_ACCESS_GRANTS_ALLOW_USERS", "True").lower() == "true"
)
USER_PERMISSIONS_CHAT_CONTROLS = (
os.environ.get("USER_PERMISSIONS_CHAT_CONTROLS", "True").lower() == "true"
@@ -1590,6 +1594,9 @@ DEFAULT_USER_PERMISSIONS = {
"notes": USER_PERMISSIONS_NOTES_ALLOW_SHARING,
"public_notes": USER_PERMISSIONS_NOTES_ALLOW_PUBLIC_SHARING,
},
"access_grants": {
"allow_users": USER_PERMISSIONS_ACCESS_GRANTS_ALLOW_USERS,
},
"chat": {
"controls": USER_PERMISSIONS_CHAT_CONTROLS,
"valves": USER_PERMISSIONS_CHAT_VALVES,
@@ -204,6 +204,33 @@ def has_public_read_access_grant(access_grants: Optional[list]) -> bool:
return False
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"] != "*":
return True
return False
def strip_user_access_grants(access_grants: Optional[list]) -> list:
"""
Remove all non-wildcard user grants from the list.
Keeps group grants and the public wildcard (user:*) intact.
"""
if not access_grants:
return []
return [
grant
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)) != "*"
)
]
def grants_to_access_control(grants: list) -> Optional[dict]:
"""
Convert a list of grant objects (AccessGrantModel or AccessGrantResponse)
+37 -1
View File
@@ -30,7 +30,7 @@ from open_webui.storage.provider import Storage
from open_webui.constants import ERROR_MESSAGES
from open_webui.utils.auth import get_verified_user, get_admin_user
from open_webui.utils.access_control import has_permission
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL
@@ -274,6 +274,18 @@ async def create_new_knowledge(
):
form_data.access_grants = []
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
knowledge = Knowledges.insert_new_knowledge(user.id, form_data)
if knowledge:
@@ -494,6 +506,18 @@ async def update_knowledge_by_id(
):
form_data.access_grants = []
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
knowledge = Knowledges.update_knowledge_by_id(id=id, form_data=form_data)
if knowledge:
# Re-embed knowledge base for semantic search
@@ -573,6 +597,18 @@ async def update_knowledge_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants("knowledge", id, form_data.access_grants, db=db)
return KnowledgeFilesResponse(
+13 -1
View File
@@ -17,7 +17,7 @@ from open_webui.models.models import (
ModelAccessResponse,
Models,
)
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from pydantic import BaseModel
from open_webui.constants import ERROR_MESSAGES
@@ -584,6 +584,18 @@ async def update_model_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants(
"model", form_data.id, form_data.access_grants, db=db
)
+25 -1
View File
@@ -28,7 +28,7 @@ from open_webui.constants import ERROR_MESSAGES
from open_webui.utils.auth import get_admin_user, get_verified_user
from open_webui.utils.access_control import has_permission
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from open_webui.internal.db import get_session
from sqlalchemy.orm import Session
@@ -296,6 +296,18 @@ async def update_note_by_id(
):
form_data.access_grants = []
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
try:
note = Notes.update_note_by_id(id, form_data, db=db)
await sio.emit(
@@ -376,6 +388,18 @@ async def update_note_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants("note", id, form_data.access_grants, db=db)
return Notes.get_note_by_id(id, db=db)
+13 -1
View File
@@ -9,7 +9,7 @@ from open_webui.models.prompts import (
PromptModel,
Prompts,
)
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from open_webui.models.groups import Groups
from open_webui.models.prompt_history import (
PromptHistories,
@@ -492,6 +492,18 @@ async def update_prompt_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants("prompt", prompt_id, form_data.access_grants, db=db)
return Prompts.get_prompt_by_id(prompt_id, db=db)
+13 -1
View File
@@ -17,7 +17,7 @@ from open_webui.models.skills import (
SkillAccessListResponse,
Skills,
)
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from open_webui.utils.auth import get_admin_user, get_verified_user
from open_webui.utils.access_control import has_access, has_permission
@@ -360,6 +360,18 @@ async def update_skill_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants("skill", id, form_data.access_grants, db=db)
return Skills.get_skill_by_id(id, db=db)
+13 -1
View File
@@ -21,7 +21,7 @@ from open_webui.models.tools import (
ToolAccessResponse,
Tools,
)
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant
from open_webui.models.access_grants import AccessGrants, has_public_read_access_grant, has_user_access_grant, strip_user_access_grants
from open_webui.utils.plugin import (
load_tool_module_by_id,
replace_imports,
@@ -595,6 +595,18 @@ async def update_tool_access_by_id(
)
]
# Strip individual user sharing if user lacks permission
if (
user.role != "admin"
and has_user_access_grant(form_data.access_grants)
and not has_permission(
user.id,
"access_grants.allow_users",
request.app.state.config.USER_PERMISSIONS,
)
):
form_data.access_grants = strip_user_access_grants(form_data.access_grants)
AccessGrants.set_access_grants("tool", id, form_data.access_grants, db=db)
return Tools.get_tool_by_id(id, db=db)
+8
View File
@@ -196,6 +196,10 @@ class SharingPermissions(BaseModel):
public_notes: bool = True
class AccessGrantsPermissions(BaseModel):
allow_users: bool = True
class ChatPermissions(BaseModel):
controls: bool = True
valves: bool = True
@@ -239,6 +243,7 @@ class SettingsPermissions(BaseModel):
class UserPermissions(BaseModel):
workspace: WorkspacePermissions
sharing: SharingPermissions
access_grants: AccessGrantsPermissions
chat: ChatPermissions
features: FeaturesPermissions
settings: SettingsPermissions
@@ -253,6 +258,9 @@ async def get_default_user_permissions(request: Request, user=Depends(get_admin_
"sharing": SharingPermissions(
**request.app.state.config.USER_PERMISSIONS.get("sharing", {})
),
"access_grants": AccessGrantsPermissions(
**request.app.state.config.USER_PERMISSIONS.get("access_grants", {})
),
"chat": ChatPermissions(
**request.app.state.config.USER_PERMISSIONS.get("chat", {})
),