engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request !2778 from rhodecode-enterprise-ce feature/RCCE-205user-that-owns-closed-PRs-cannot-be-deleted-as-PRs-cannot-be-deleted-or-unclosed

Browse source 
Changes from branch: Feature/RCCE 205user that owns closed PRs cannot be deleted as PRs cannot be deleted or unclosed
This commit is contained in:
Andrii Verbytskyi 2025-08-01 09:30:32 +00:00
commit 58fce87f39
27 changed files with 6033 additions and 386 deletions
Download patch file Download diff file Expand all files Collapse all files

2
rhodecode/__init__.py
View file

@ -100,7 +100,7 @@ PYRAMID_SETTINGS = {}
EXTENSIONS = {}
__version__ = ".".join((str(each) for each in VERSION[:3]))
__dbversion__ = 116 # defines current db version for migrations
__dbversion__ = 117 # defines current db version for migrations
__license__ = "AGPLv3, and Commercial License"
__author__ = "RhodeCode GmbH"
__url__ = "https://code.rhodecode.com"

7
rhodecode/api/tests/test_get_users.py
View file

@ -31,7 +31,12 @@ class TestGetUsers(object):
)
response = api_call(self.app, params)
ret_all = []
_users = User.query().filter(User.username != User.DEFAULT_USER).order_by(User.username).all()
_users = (
User.query()
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
.order_by(User.username)
.all()
)
for usr in _users:
ret = usr.get_api_data(include_secrets=True)
ret_all.append(jsonify(ret))

4
rhodecode/api/views/user_api.py
View file

@ -145,7 +145,9 @@ def get_users(request, apiuser):
raise JSONRPCForbidden()
result = []
users_list = User.query().order_by(User.username).filter(User.username != User.DEFAULT_USER).all()
users_list = (
User.query().order_by(User.username).filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).all()
)
for user in users_list:
result.append(user.get_api_data(include_secrets=True))
return result

8
rhodecode/apps/_base/__init__.py
View file

@ -144,7 +144,7 @@ class BaseAppView(object):
if not user_obj:
return
if user_obj.username == User.DEFAULT_USER:
if user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
return
now = time.time()
@ -270,7 +270,7 @@ class RepoAppView(BaseAppView):
def _prepare_and_set_clone_url(self, c):
username = ""
if self._rhodecode_user.username != User.DEFAULT_USER:
if self._rhodecode_user.username in [User.DEFAULT_USER, User.GHOST_USER]:
username = self._rhodecode_user.username
_def_clone_uri = c.clone_uri_tmpl
@ -558,9 +558,9 @@ class UserAppView(BaseAppView):
_ = self.request.translate
if not request.db_user_supports_default:
if self.db_user.username == User.DEFAULT_USER:
if self.db_user.username in [User.DEFAULT_USER, User.GHOST_USER]:
h.flash(
_("Editing user `{}` is disabled.".format(User.DEFAULT_USER)),
_("Editing user `{}` is disabled.".format(self.db_user.username)),
category="warning",
)
raise HTTPFound(h.route_path("users"))

138
rhodecode/apps/admin/tests/test_admin_users.py
View file

@ -21,7 +21,7 @@ from sqlalchemy.orm.exc import NoResultFound
from rhodecode.lib import auth
from rhodecode.lib import helpers as h
from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository
from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository, RepoGroup, UserGroup
from rhodecode.model.meta import Session
from rhodecode.model.user import UserModel
@ -41,7 +41,7 @@ class TestAdminUsersView(TestController):
self.log_user()
response = self.app.get(route_path("users_data"), extra_environ=xhr_header)
all_users = User.query().filter(User.username != User.DEFAULT_USER).count()
all_users = User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).count()
assert response.json["recordsTotal"] == all_users
def test_show_users_data_filtered(self, xhr_header):
@ -50,7 +50,7 @@ class TestAdminUsersView(TestController):
route_path("users_data", params={"search[value]": "empty_search"}), extra_environ=xhr_header
)
all_users = User.query().filter(User.username != User.DEFAULT_USER).count()
all_users = User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).count()
assert response.json["recordsTotal"] == all_users
assert response.json["recordsFiltered"] == 0
@ -448,25 +448,6 @@ class TestAdminUsersView(TestController):
assert_session_flash(response, "Successfully deleted user `{}`".format(username))
def test_delete_owner_of_repository(self, request, user_util):
self.log_user()
obj_name = "test_repo"
usr = user_util.create_user()
username = usr.username
fixture.create_repo(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 repositories and cannot be removed. '
"Switch owners or remove those repositories:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_repo(obj_name)
def test_delete_owner_of_repository_detaching(self, request, user_util):
self.log_user()
obj_name = "test_repo"
@ -478,48 +459,18 @@ class TestAdminUsersView(TestController):
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repos": "detach", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Detached 1 repositories"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
repo = Repository.get_by_repo_name(obj_name)
assert repo.user_id == ghost_user.user_id, "Ghost user should be a new owner of the repo."
fixture.destroy_repo(obj_name)
def test_delete_owner_of_repository_deleting(self, request, user_util):
self.log_user()
obj_name = "test_repo"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_repo(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repos": "delete", "csrf_token": self.csrf_token},
)
msg = "Deleted 1 repositories"
assert_session_flash(response, msg)
def test_delete_owner_of_repository_group(self, request, user_util):
self.log_user()
obj_name = "test_group"
usr = user_util.create_user()
username = usr.username
fixture.create_repo_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 repository groups and cannot be removed. '
"Switch owners or remove those repository groups:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_repo_group(obj_name)
def test_delete_owner_of_repository_group_detaching(self, request, user_util):
self.log_user()
obj_name = "test_group"
@ -530,48 +481,18 @@ class TestAdminUsersView(TestController):
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repo_groups": "delete", "csrf_token": self.csrf_token},
)
msg = "Deleted 1 repository groups"
assert_session_flash(response, msg)
def test_delete_owner_of_repository_group_deleting(self, request, user_util):
self.log_user()
obj_name = "test_group"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_repo_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repo_groups": "detach", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Detached 1 repository groups"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
repo_group = RepoGroup.get_by_group_name(obj_name)
assert repo_group.user_id == ghost_user.user_id, "Ghost user should be a new owner of the repo_group."
fixture.destroy_repo_group(obj_name)
def test_delete_owner_of_user_group(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
usr = user_util.create_user()
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 user groups and cannot be removed. '
"Switch owners or remove those user groups:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_user_group(obj_name)
def test_delete_owner_of_user_group_detaching(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
@ -579,34 +500,21 @@ class TestAdminUsersView(TestController):
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
try:
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_user_groups": "detach", "csrf_token": self.csrf_token},
)
msg = "Detached 1 user groups"
assert_session_flash(response, msg)
finally:
fixture.destroy_user_group(obj_name)
def test_delete_owner_of_user_group_deleting(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_user_groups": "delete", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Deleted 1 user groups"
msg = "Detached 1 user groups"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
user_group = UserGroup.get_by_group_name(obj_name)
assert user_group.user_id == ghost_user.user_id, "Ghost user should be a new owner of the user_group."
fixture.destroy_user_group(obj_name)
def test_edit(self, user_util):
self.log_user()
user = user_util.create_user()

105
rhodecode/apps/admin/views/users.py
View file

@ -104,14 +104,13 @@ class AdminUsersView(BaseAppView, DataGridAppView):
def user_actions(user_id, username):
return _render("user_actions", user_id, username)
users_data_total_count = User.query().filter(User.username != User.DEFAULT_USER).count()
excluded_users = User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])
users_data_total_count = User.query().filter(excluded_users).count()
users_data_total_inactive_count = (
User.query().filter(User.username != User.DEFAULT_USER).filter(User.active != true()).count()
)
users_data_total_inactive_count = User.query().filter(excluded_users).filter(User.active != true()).count()
# json generate
base_q = User.query().filter(User.username != User.DEFAULT_USER)
base_q = User.query().filter(excluded_users)
base_inactive_q = base_q.filter(User.active != true())
if search_q:
@ -366,91 +365,61 @@ class UsersView(UserAppView):
_user_groups = len(c.user.user_groups)
_pull_requests = len(c.user.user_pull_requests)
_artifacts = len(c.user.artifacts)
_reviewer = len(c.user.reviewer_pull_requests)
handle_repos = None
handle_repo_groups = None
handle_user_groups = None
handle_pull_requests = None
handle_artifacts = None
# calls for flash of handle based on handle case detach or delete
def set_handle_flash_repos():
handle = handle_repos
if handle == "detach":
def set_flash_repos_if_detached():
if detach_repos:
h.flash(_("Detached %s repositories") % _repos, category="success")
elif handle == "delete":
h.flash(_("Deleted %s repositories") % _repos, category="success")
def set_handle_flash_repo_groups():
handle = handle_repo_groups
if handle == "detach":
def set_flash_reviews_if_deleted():
if delete_reviewer:
h.flash(_("Removed user from %s pull requests as a reviewer\\observer") % _reviewer, category="success")
def set_flash_repo_groups_if_detached():
if detach_repo_groups:
h.flash(_("Detached %s repository groups") % _repo_groups, category="success")
elif handle == "delete":
h.flash(_("Deleted %s repository groups") % _repo_groups, category="success")
def set_handle_flash_user_groups():
handle = handle_user_groups
if handle == "detach":
def set_flash_user_groups_if_detached():
if detach_user_groups:
h.flash(_("Detached %s user groups") % _user_groups, category="success")
elif handle == "delete":
h.flash(_("Deleted %s user groups") % _user_groups, category="success")
def set_handle_flash_pull_requests():
handle = handle_pull_requests
if handle == "detach":
def set_flash_pull_requests_if_detached():
if detach_pull_requests:
h.flash(_("Detached %s pull requests") % _pull_requests, category="success")
elif handle == "delete":
h.flash(_("Deleted %s pull requests") % _pull_requests, category="success")
def set_handle_flash_artifacts():
handle = handle_artifacts
if handle == "detach":
def set_flash_artifacts_if_detached():
if detach_artifacts:
h.flash(_("Detached %s artifacts") % _artifacts, category="success")
elif handle == "delete":
h.flash(_("Deleted %s artifacts") % _artifacts, category="success")
handle_user = User.get_first_super_admin()
handle_user_id = safe_int(self.request.POST.get("detach_user_id"))
if handle_user_id:
# NOTE(marcink): we get new owner for objects...
handle_user = User.get_or_404(handle_user_id)
if _repos and self.request.POST.get("user_repos"):
handle_repos = self.request.POST["user_repos"]
if _repo_groups and self.request.POST.get("user_repo_groups"):
handle_repo_groups = self.request.POST["user_repo_groups"]
if _user_groups and self.request.POST.get("user_user_groups"):
handle_user_groups = self.request.POST["user_user_groups"]
if _pull_requests and self.request.POST.get("user_pull_requests"):
handle_pull_requests = self.request.POST["user_pull_requests"]
if _artifacts and self.request.POST.get("user_artifacts"):
handle_artifacts = self.request.POST["user_artifacts"]
detach_repos = _repos > 0
detach_repo_groups = _repo_groups > 0
detach_user_groups = _user_groups > 0
detach_pull_requests = _pull_requests > 0
detach_artifacts = _artifacts > 0
delete_reviewer = _reviewer > 0
old_values = c.user.get_api_data()
try:
UserModel().delete(
c.user,
handle_repos=handle_repos,
handle_repo_groups=handle_repo_groups,
handle_user_groups=handle_user_groups,
handle_pull_requests=handle_pull_requests,
handle_artifacts=handle_artifacts,
handle_new_owner=handle_user,
detach_repos=detach_repos,
detach_repo_groups=detach_repo_groups,
detach_user_groups=detach_user_groups,
detach_pull_requests=detach_pull_requests,
detach_artifacts=detach_artifacts,
delete_reviewer=delete_reviewer,
)
audit_logger.store_web("user.delete", action_data={"old_data": old_values}, user=c.rhodecode_user)
Session().commit()
set_handle_flash_repos()
set_handle_flash_repo_groups()
set_handle_flash_user_groups()
set_handle_flash_pull_requests()
set_handle_flash_artifacts()
set_flash_repos_if_detached()
set_flash_repo_groups_if_detached()
set_flash_user_groups_if_detached()
set_flash_pull_requests_if_detached()
set_flash_artifacts_if_detached()
set_flash_reviews_if_deleted()
username = h.escape(old_values["username"])
h.flash(_("Successfully deleted user `{}`").format(username), category="success")
except (

8
rhodecode/apps/home/views.py
View file

@ -78,12 +78,12 @@ class HomeView(BaseAppView, DataGridAppView):
_users = UserModel().get_users(name_contains=query, only_active=active)
def maybe_skip_default_user(usr):
if skip_default_user and usr["username"] == UserModel.cls.DEFAULT_USER:
def maybe_skip_user(usr):
if skip_default_user and usr["username"] in [UserModel.cls.DEFAULT_USER, UserModel.cls.GHOST_USER]:
return False
return True
_users = list(filter(maybe_skip_default_user, _users))
_users = list(filter(maybe_skip_user, _users))
if include_groups:
# extend with user groups
@ -239,7 +239,7 @@ class HomeView(BaseAppView, DataGridAppView):
User.query()
.order_by(func.length(User.username))
.order_by(User.username)
.filter(User.username != User.DEFAULT_USER)
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
)
if name_contains:

2
rhodecode/apps/user_group_profile/views.py
View file

@ -46,5 +46,5 @@ class UserGroupProfileView(BaseAppView):
raise HTTPNotFound()
group_members_obj = sorted((x.user for x in c.user_group.members), key=lambda u: u.username.lower())
c.group_members = group_members_obj
c.anonymous = self._rhodecode_user.username == User.DEFAULT_USER
c.anonymous = self._rhodecode_user.username in [User.DEFAULT_USER, User.GHOST_USER]
return self._get_template_context(c)

3
rhodecode/apps/user_profile/tests/test_users.py
View file

@ -67,5 +67,8 @@ class TestUsersController(TestController):
# default user, not allowed to show it
self.app.get(route_path("user_profile", username=User.DEFAULT_USER), status=404)
# ghost user, not allowed to show it
self.app.get(route_path("user_profile", username=User.GHOST_USER), status=404)
# actual 404
self.app.get(route_path("user_profile", username="unknown"), status=404)

2
rhodecode/apps/user_profile/views.py
View file

@ -40,7 +40,7 @@ class UserProfileView(BaseAppView):
username = self.request.matchdict.get("username")
c.user = UserModel().get_by_username(username)
if not c.user or c.user.username == User.DEFAULT_USER:
if not c.user or c.user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise HTTPNotFound()
return self._get_template_context(c)

4
rhodecode/authentication/plugins/auth_rhodecode.py
View file

@ -138,6 +138,10 @@ class RhodeCodeAuthPlugin(RhodeCodeAuthPluginBase):
log.warning("userobj:%s is not super-admin and auth restriction is set to %s", userobj, auth_restriction)
return None
if userobj.username == User.GHOST_USER:
log.warning("userobj:%s is not allowed to login", userobj)
return None
user_attrs = {
"username": userobj.username,
"firstname": userobj.firstname,

4
rhodecode/lib/audit_logger.py
View file

@ -27,6 +27,9 @@ from rhodecode.lib.str_utils import safe_str
log = logging.getLogger(__name__)
PR_IDS_REVIEWER_ROLE = "pull_request_ids_reviewer_role"
PR_IDS_OBSERVER_ROLE = "pull_request_ids_observer_role"
# action as key, and expected action_data as value
ACTIONS_V1 = {
"user.login.success": {"user_agent": ""},
@ -63,6 +66,7 @@ ACTIONS_V1 = {
"repo.edit.permissions.branch": {},
"repo.archive": {"old_data": {}},
"repo.delete": {"old_data": {}},
"review.delete": {PR_IDS_OBSERVER_ROLE: [], PR_IDS_REVIEWER_ROLE: [], "prev_user": {}},
"repo.archive.download": {"user_agent": "", "archive_name": "", "archive_spec": "", "archive_cached": ""},
"repo.permissions.branch_rule.create": {},
"repo.permissions.branch_rule.edit": {},

8
rhodecode/lib/auth.py
View file

@ -1295,6 +1295,10 @@ class AuthUser(object):
def is_default(self):
return self.username == User.DEFAULT_USER
@property
def is_ghost(self):
return self.username == User.GHOST_USER
@property
def is_admin(self):
return self.admin
@ -1878,7 +1882,7 @@ class NotAnonymous(object):
_ = request.translate
log.debug("Checking if user is not anonymous @%s", cls)
anonymous = self.user.username == User.DEFAULT_USER
anonymous = self.user.username in [User.DEFAULT_USER, User.GHOST_USER]
if anonymous:
came_from = get_came_from(request)
@ -1918,7 +1922,7 @@ class PermsDecorator(object):
else:
log.debug("Permission denied for %s %s", cls, _user)
anonymous = _user.username == User.DEFAULT_USER
anonymous = _user.username in [User.DEFAULT_USER, User.GHOST_USER]
if anonymous:
import rhodecode.lib.helpers as h

15
rhodecode/lib/db_manage.py
View file

@ -605,6 +605,21 @@ class DbManage(object):
user=user, lifetime_minutes=-1, role=UserModel.auth_token_role.ROLE_ALL, description="BUILTIN TOKEN"
)
def create_ghost_user(self):
if not User.get_by_username(User.GHOST_USER):
log.info("creating ghost user")
ghost_user = User()
ghost_user.name = User.GHOST_USER.capitalize()
ghost_user.username = User.GHOST_USER
ghost_user.lastname = "User"
ghost_user.strict_creation_check = False
ghost_user.email = User.GHOST_USER_EMAIL
ghost_user.active = False
ghost_user.description = "System user to own assets of deleted users (does not consume license slot)"
ghost_user.is_new_user = False
Session().add(ghost_user)
Session().commit()
def create_default_user(self):
log.info("creating default user")
# create default user for handling default permissions.

5678
rhodecode/lib/dbmigrate/schema/db_4_20_0_1.py Normal file
View file

File diff suppressed because it is too large Load diff

47
rhodecode/lib/dbmigrate/versions/117_version_5_7_0.py Normal file
View file

@ -0,0 +1,47 @@
import json
import logging
from psycopg2.errorcodes import DUPLICATE_COLUMN
from sqlalchemy import *
from sqlalchemy.engine import reflection
from alembic.migration import MigrationContext
from alembic.operations import Operations
from rhodecode.lib.dbmigrate.versions import _reset_base
from rhodecode.lib.jsonalchemy import MutationObj, JsonType
from rhodecode.model import meta, init_model_encryption
from rhodecode.model.db import User
def upgrade(migrate_engine):
"""
Upgrade operations go here.
Don't create your own engine; bind migrate_engine to your metadata
"""
_reset_base(migrate_engine)
from rhodecode.lib.dbmigrate.schema import db_4_20_0_1 as db
init_model_encryption(db)
UserModel: User = db.User
if not User.get_by_username(UserModel.GHOST_USER):
ghost_user = UserModel()
ghost_user.name = UserModel.GHOST_USER.capitalize()
ghost_user.username = UserModel.GHOST_USER
ghost_user.lastname = "User"
ghost_user.strict_creation_check = False
ghost_user.email = UserModel.GHOST_USER_EMAIL
ghost_user.active = False
ghost_user.description = "System user to own assets of deleted users (does not consume license slot)"
ghost_user.is_new_user = False
session = meta.Session()
session.add(ghost_user)
session.commit()
def downgrade(migrate_engine):
pass

2
rhodecode/lib/helpers.py
View file

@ -1035,7 +1035,7 @@ def link_to_user(author, length=0, **kwargs):
if length:
display_person = shorter(display_person, length)
if user and user.username != user.DEFAULT_USER:
if user and user.username not in [user.DEFAULT_USER, user.GHOST_USER]:
return link_to(escape(display_person), route_path("user_profile", username=user.username), **kwargs)
else:
return escape(display_person)

2
rhodecode/lib/middleware/simplevcs.py
View file

@ -484,7 +484,7 @@ class SimpleVCS(object):
auth_user = user_obj.AuthUser()
if (
user_obj
and user_obj.username != User.DEFAULT_USER
and user_obj.username not in [User.DEFAULT_USER, User.GHOST_USER]
and user_obj.user_data.get("force_password_change")
):
reason = "password change required"

1
rhodecode/lib/rc_commands/setup_rc.py
View file

@ -96,6 +96,7 @@ def command(
opts = dbmanage.config_prompt(None)
dbmanage.create_settings(opts)
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_admin_and_prompt()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()

8
rhodecode/lib/utils.py
View file

@ -536,9 +536,6 @@ def get_rhodecode_repo_store_path():
Returns the base path. The base path is the filesystem path which points
to the repository store.
"""
import rhodecode
return rhodecode.CONFIG["repo_store.path"]
@ -756,7 +753,6 @@ def deep_reload_package(package_name):
def load_rcextensions(root_path):
import rhodecode
from rhodecode.config import conf
path = os.path.join(root_path)
@ -785,7 +781,6 @@ def get_custom_lexer(extension):
returns a custom lexer if it is defined in rcextensions module, or None
if there's no custom lexer defined
"""
import rhodecode
from pygments import lexers
# custom override made by RhodeCode
@ -846,6 +841,7 @@ def create_test_database(test_path, config):
# for tests dynamically set new root paths based on generated content
dbmanage.create_settings(dbmanage.config_prompt(test_path))
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_test_admin_and_users()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()
@ -881,7 +877,7 @@ def create_test_repositories(test_path, config):
def password_changed(auth_user, session):
# Never report password change in case of default user or anonymous user.
if auth_user.username == User.DEFAULT_USER or auth_user.user_id is None:
if auth_user.username in [User.DEFAULT_USER, User.GHOST_USER] or auth_user.user_id is None:
return False
password_hash = md5(safe_bytes(auth_user.password)) if auth_user.password else None

22
rhodecode/model/db.py
View file

@ -134,11 +134,11 @@ PERMISSION_TYPE_SORT = {
def display_user_sort(obj):
"""
Sort function used to sort permissions in .permissions() function of
Repository, RepoGroup, UserGroup. Also it put the default user in front
Repository, RepoGroup, UserGroup. Also it put the default and ghost users in front
of all other resources
"""
if obj.username == User.DEFAULT_USER:
if obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
return "#####"
prefix = PERMISSION_TYPE_SORT.get(obj.permission.split(".")[-1], "")
extra_sort_num = "1" # default
@ -617,6 +617,8 @@ class User(Base, BaseModel):
DEFAULT_USER = "default"
DEFAULT_USER_EMAIL = "anonymous@rhodecode.org"
GHOST_USER = "ghost"
GHOST_USER_EMAIL = "ghost@rhodecode.org"
DEFAULT_GRAVATAR_URL = "https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}"
RECOVERY_CODES_COUNT = 10
@ -1280,13 +1282,21 @@ class User(Base, BaseModel):
qry = qry.filter(User.active == true())
return [x.user_id for x in qry]
@classmethod
def get_ghost_user(cls, cache=False, refresh=False):
return cls._get_system_user(username=cls.GHOST_USER, cache=cache, refresh=refresh)
@classmethod
def get_default_user(cls, cache=False, refresh=False):
user = User.get_by_username(User.DEFAULT_USER, cache=cache)
return cls._get_system_user(username=cls.DEFAULT_USER, cache=cache, refresh=refresh)
@classmethod
def _get_system_user(cls, username, cache=False, refresh=False):
user = cls.get_by_username(username, cache=cache)
if user is None:
raise Exception("FATAL: Missing default account!")
raise Exception("FATAL: Missing %s account!" % username)
if refresh:
# The default user might be based on outdated state which
# The 'system' user might be based on outdated state which
# has been loaded from the cache.
# A call to refresh() ensures that the
# latest state from the database is used.
@ -3037,7 +3047,7 @@ class RepoGroup(Base, BaseModel):
@classmethod
def get_user_personal_repo_group(cls, user_id):
user = User.get(user_id)
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
return None
return cls.query().filter(cls.personal == true()).filter(cls.user == user).order_by(cls.group_id.asc()).first()

4
rhodecode/model/repo.py
View file

@ -664,6 +664,10 @@ class RepoModel(BaseModel):
member_id = int(member_id)
if member_type == "user":
member_name = User.get(member_id).username
if member_name == User.GHOST_USER:
raise ValueError("Modify user permissions for ghost user is not allowed.")
if member_name == User.DEFAULT_USER:
# NOTE(dan): detect if we changed permissions for default user
perm_obj = (

10
rhodecode/model/repo_group.py
View file

@ -359,7 +359,7 @@ class RepoGroupModel(BaseModel):
elif isinstance(_obj, Repository):
# private repos will not allow to change the default
# permissions using recursive mode
if _obj.private and _user_obj.username == User.DEFAULT_USER:
if _obj.private and _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
@ -381,7 +381,7 @@ class RepoGroupModel(BaseModel):
elif isinstance(_obj, Repository):
# private repos will not allow to change the default
# permissions using recursive mode, also there's no revocation fo default user, just update
if _user_obj.username == User.DEFAULT_USER:
if _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
RepoModel().revoke_user_permission(repo=_obj, user=_user_obj)
@ -425,7 +425,11 @@ class RepoGroupModel(BaseModel):
if member_type == "user":
member_obj = User.get(member_id)
member_name = member_obj.username
if isinstance(obj, RepoGroup) and obj == repo_group and member_name == User.DEFAULT_USER:
is_repo_group = isinstance(obj, RepoGroup) and obj == repo_group
if is_repo_group and member_name == User.GHOST_USER:
raise ValueError("Modify user permissions for ghost user is not allowed.")
if is_repo_group and member_name == User.DEFAULT_USER:
# NOTE(dan): detect if we changed permissions for default user
perm_obj = (
self.sa.query(UserRepoGroupToPerm)

278
rhodecode/model/user.py
View file

@ -24,10 +24,12 @@ import logging
import traceback
import datetime
import ipaddress
from typing import List
from sqlalchemy.exc import DatabaseError
from rhodecode import events
from rhodecode.lib import audit_logger
from rhodecode.lib.pyramid_utils import get_current_request
from rhodecode.lib.user_log_filter import user_log_filter
from rhodecode.lib.utils2 import get_current_rhodecode_user, action_logger_generic, AttributeDict, str2bool
@ -56,6 +58,7 @@ from rhodecode.model.db import (
UserEmailMap,
UserIpMap,
UserLog,
PullRequestReviewers,
)
from rhodecode.model.meta import Session
from rhodecode.model.auth_token import AuthTokenModel
@ -128,6 +131,9 @@ class UserModel(BaseModel):
user = user.options(FromCache("sql_cache_short", f"get_user_{name_key}"))
return user.scalar()
def get_by_id(self, id_):
return self.sa.query(User).filter(User.user_id == id_).scalar()
def get_by_email(self, email, cache=False, case_insensitive=False):
return User.get_by_email(email, case_insensitive, cache)
@ -135,7 +141,11 @@ class UserModel(BaseModel):
return User.get_by_auth_token(auth_token, cache)
def get_active_user_count(self, cache=False):
qry = User.query().filter(User.active == true()).filter(User.username != User.DEFAULT_USER)
qry = (
User.query()
.filter(User.active == true())
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
)
if cache:
qry = qry.options(FromCache("sql_cache_short", "get_active_users"))
return qry.count()
@ -169,10 +179,8 @@ class UserModel(BaseModel):
raise
def update_user(self, user, skip_attrs=None, **kwargs):
from rhodecode.lib.auth import get_crypt_password
user = self._get_user(user)
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -356,7 +364,7 @@ class UserModel(BaseModel):
edit = True
# we're not allowed to edit default user
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -459,131 +467,106 @@ class UserModel(BaseModel):
log.error(traceback.format_exc())
raise
def _handle_user_repos(self, username, repositories, handle_user, handle_mode=None):
left_overs = True
def _detach_user_repos(self, username, repositories, new_owner):
for obj in repositories:
obj.user = new_owner
# set description we know why ghost now owns
# additional repositories that were orphaned !
obj.description += " \n::detached repository from deleted user: %s" % (username,)
self.sa.add(obj)
from rhodecode.model.repo import RepoModel
def _detach_user_repo_groups(self, username, repository_groups, new_owner):
for r in repository_groups:
r.user = new_owner
# set description we know why ghost now owns
# additional repositories that were orphaned !
r.group_description += " \n::detached repository group from deleted user: %s" % (username,)
r.personal = False
self.sa.add(r)
if handle_mode == "detach":
for obj in repositories:
obj.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
obj.description += " \n::detached repository from deleted user: %s" % (username,)
self.sa.add(obj)
left_overs = False
elif handle_mode == "delete":
for obj in repositories:
RepoModel().delete(obj, forks="detach")
left_overs = False
def _detach_user_user_groups(self, username, user_groups, new_owner):
for r in user_groups:
for user_user_group_to_perm in r.user_user_group_to_perm:
if user_user_group_to_perm.user.username == username:
user_user_group_to_perm.user = new_owner
r.user = new_owner
# set description we know why ghost now owns
# additional repositories that were orphaned !
r.user_group_description += " \n::detached user group from deleted user: %s" % (username,)
self.sa.add(r)
# if nothing is done we have left overs left
return left_overs
def _detach_user_pull_requests(self, username, pull_requests, new_owner):
for pr in pull_requests:
pr.user_id = new_owner.user_id
# set description we know why ghost now owns
# additional repositories that were orphaned !
pr.description += " \n::detached pull requests from deleted user: %s" % (username,)
self.sa.add(pr)
def _handle_user_repo_groups(self, username, repository_groups, handle_user, handle_mode=None):
left_overs = True
def _detach_user_artifacts(self, username, artifacts, new_owner):
for a in artifacts:
a.upload_user = new_owner
# set description we know why ghost now owns
# additional artifacts that were orphaned !
a.file_description += " \n::detached artifact from deleted user: %s" % (username,)
self.sa.add(a)
from rhodecode.model.repo_group import RepoGroupModel
def _delete_from_reviews(self, user: User, reviewer_pull_requests: List[PullRequestReviewers]):
repos_map = {}
if handle_mode == "detach":
for r in repository_groups:
r.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
r.group_description += " \n::detached repository group from deleted user: %s" % (username,)
r.personal = False
self.sa.add(r)
left_overs = False
elif handle_mode == "delete":
for r in repository_groups:
RepoGroupModel().delete(r)
left_overs = False
for review in reviewer_pull_requests:
repo_id = review.pull_request.target_repo_id
repo_map = repos_map.get(repo_id, {})
# if nothing is done we have left overs left
return left_overs
pull_request_ids_reviewer_role = repo_map.get(audit_logger.PR_IDS_REVIEWER_ROLE, set())
pull_request_ids_observer_role = repo_map.get(audit_logger.PR_IDS_OBSERVER_ROLE, set())
def _handle_user_user_groups(self, username, user_groups, handle_user, handle_mode=None):
left_overs = True
if review.role == PullRequestReviewers.ROLE_REVIEWER:
pull_request_ids_reviewer_role.add(review.pull_request.pull_request_id)
elif review.role == PullRequestReviewers.ROLE_OBSERVER:
pull_request_ids_observer_role.add(review.pull_request.pull_request_id)
from rhodecode.model.user_group import UserGroupModel
repos_map[repo_id] = {
audit_logger.PR_IDS_REVIEWER_ROLE: pull_request_ids_reviewer_role,
audit_logger.PR_IDS_OBSERVER_ROLE: pull_request_ids_observer_role,
"repo_id": repo_id,
"repo_name": review.pull_request.target_repo.repo_name,
}
if handle_mode == "detach":
for r in user_groups:
for user_user_group_to_perm in r.user_user_group_to_perm:
if user_user_group_to_perm.user.username == username:
user_user_group_to_perm.user = handle_user
r.user = handle_user
# set description we know why we super admin now owns
# additional repositories that were orphaned !
r.user_group_description += " \n::detached user group from deleted user: %s" % (username,)
self.sa.add(r)
left_overs = False
elif handle_mode == "delete":
for r in user_groups:
UserGroupModel().delete(r)
left_overs = False
self.sa.delete(review)
# if nothing is done we have left overs left
return left_overs
for repo_id in repos_map:
repo_map = repos_map[repo_id]
def _handle_user_pull_requests(self, username, pull_requests, handle_user, handle_mode=None):
left_overs = True
from rhodecode.model.pull_request import PullRequestModel
if handle_mode == "detach":
for pr in pull_requests:
pr.user_id = handle_user.user_id
# set description we know why we super admin now owns
# additional repositories that were orphaned !
pr.description += " \n::detached pull requests from deleted user: %s" % (username,)
self.sa.add(pr)
left_overs = False
elif handle_mode == "delete":
for pr in pull_requests:
PullRequestModel().delete(pr)
left_overs = False
# if nothing is done we have leftovers left
return left_overs
def _handle_user_artifacts(self, username, artifacts, handle_user, handle_mode=None):
left_overs = True
if handle_mode == "detach":
for a in artifacts:
a.upload_user = handle_user
# set description we know why we super admin now owns
# additional artifacts that were orphaned !
a.file_description += " \n::detached artifact from deleted user: %s" % (username,)
self.sa.add(a)
left_overs = False
elif handle_mode == "delete":
from rhodecode.apps.file_store import utils as store_utils
request = get_current_request()
f_store = store_utils.get_filestore_backend(request.registry.settings)
for a in artifacts:
file_uid = a.file_uid
f_store.delete(file_uid)
self.sa.delete(a)
left_overs = False
# if nothing is done we have left overs left
return left_overs
audit_logger.store_web(
"review.delete",
user=get_current_rhodecode_user(),
action_data={
audit_logger.PR_IDS_OBSERVER_ROLE: list(repo_map.get(audit_logger.PR_IDS_OBSERVER_ROLE, set())),
audit_logger.PR_IDS_REVIEWER_ROLE: list(repo_map.get(audit_logger.PR_IDS_REVIEWER_ROLE, set())),
"prev_user": {
"username": user.username,
"name": user.name,
"email": user.email,
},
},
repo=audit_logger.RepoWrap(
repo_id=repo_map["repo_id"],
repo_name=repo_map["repo_name"],
),
)
def delete(
self,
user,
cur_user=None,
handle_repos=None,
handle_repo_groups=None,
handle_user_groups=None,
handle_pull_requests=None,
handle_artifacts=None,
handle_new_owner=None,
detach_repos=False,
detach_repo_groups=False,
detach_user_groups=False,
detach_pull_requests=False,
detach_artifacts=False,
delete_reviewer=False,
new_owner=None,
):
from rhodecode.lib import hooks_base
@ -593,63 +576,30 @@ class UserModel(BaseModel):
user = self._get_user(user)
try:
if user.username == User.DEFAULT_USER:
if user.username in [User.DEFAULT_USER, User.GHOST_USER]:
raise DefaultUserException("You can't remove this user since it's crucial for entire application")
handle_user = handle_new_owner or self.cls.get_first_super_admin()
log.debug("New detached objects owner %s", handle_user)
left_overs = self._handle_user_repos(user.username, user.repositories, handle_user, handle_repos)
if left_overs and user.repositories:
repos = [x.repo_name for x in user.repositories]
raise UserOwnsReposException(
'user "%(username)s" still owns %(len_repos)s repositories and cannot be '
"removed. Switch owners or remove those repositories:%(list_repos)s"
% {"username": user.username, "len_repos": len(repos), "list_repos": ", ".join(repos)}
)
if delete_reviewer:
self._delete_from_reviews(user, user.reviewer_pull_requests)
left_overs = self._handle_user_repo_groups(
user.username, user.repository_groups, handle_user, handle_repo_groups
)
if left_overs and user.repository_groups:
repo_groups = [x.group_name for x in user.repository_groups]
raise UserOwnsRepoGroupsException(
'user "%(username)s" still owns %(len_repo_groups)s repository groups and cannot be '
"removed. Switch owners or remove those repository groups:%(list_repo_groups)s"
% {
"username": user.username,
"len_repo_groups": len(repo_groups),
"list_repo_groups": ", ".join(repo_groups),
}
)
if detach_repos or detach_repo_groups or detach_user_groups or detach_pull_requests or detach_artifacts:
new_owner = new_owner or self.cls.get_ghost_user()
log.debug("New detached objects owner %s", new_owner)
left_overs = self._handle_user_user_groups(user.username, user.user_groups, handle_user, handle_user_groups)
if left_overs and user.user_groups:
user_groups = [x.users_group_name for x in user.user_groups]
raise UserOwnsUserGroupsException(
'user "%s" still owns %s user groups and cannot be '
"removed. Switch owners or remove those user groups:%s"
% (user.username, len(user_groups), ", ".join(user_groups))
)
if detach_repos:
self._detach_user_repos(user.username, user.repositories, new_owner)
left_overs = self._handle_user_pull_requests(
user.username, user.user_pull_requests, handle_user, handle_pull_requests
)
if left_overs and user.user_pull_requests:
pull_requests = [f"!{x.pull_request_id}" for x in user.user_pull_requests]
raise UserOwnsPullRequestsException(
'user "%s" still owns %s pull requests and cannot be '
"removed. Switch owners or remove those pull requests:%s"
% (user.username, len(pull_requests), ", ".join(pull_requests))
)
if detach_repo_groups:
self._detach_user_repo_groups(user.username, user.repository_groups, new_owner)
left_overs = self._handle_user_artifacts(user.username, user.artifacts, handle_user, handle_artifacts)
if left_overs and user.artifacts:
artifacts = [x.file_uid for x in user.artifacts]
raise UserOwnsArtifactsException(
'user "%s" still owns %s artifacts and cannot be '
"removed. Switch owners or remove those artifacts:%s"
% (user.username, len(artifacts), ", ".join(artifacts))
)
if detach_user_groups:
self._detach_user_user_groups(user.username, user.user_groups, new_owner)
if detach_pull_requests:
self._detach_user_pull_requests(user.username, user.user_pull_requests, new_owner)
if detach_artifacts:
self._detach_user_artifacts(user.username, user.artifacts, new_owner)
user_data = user.get_dict() # fetch user data before expire

4
rhodecode/tests/fixtures/fixture_utils.py vendored
View file

@ -1297,8 +1297,10 @@ class UserUtility(object):
self.fixture.destroy_user_group(user_group_id)
def _cleanup_users(self):
user_model = UserModel()
for user_id in self.user_ids:
self.fixture.destroy_user(user_id)
if user_model.get_by_id(user_id):
self.fixture.destroy_user(user_id)
@pytest.fixture(scope="session")

36
rhodecode/tests/lib/test_auth.py
View file

@ -572,7 +572,12 @@ def test_auth_user_get_cookie_store_for_normal_user(user_util):
assert auth_user.get_cookie_store() == expected_data
def test_auth_user_get_cookie_store_for_default_user():
def test_auth_user_get_cookie_store_for_default_user(baseapp):
# NOTE: Importing baseapp is important—if this test runs in isolation,
# the User class may not be bound to the Session, resulting in an error.
# Importing baseapp ensures the session is explicitly bound for this test.
# When tests are run in batch, this binding often happens implicitly as a side effect of other tests or fixtures.
default_user = User.get_default_user()
auth_user = auth.AuthUser()
expected_data = {
@ -584,6 +589,35 @@ def test_auth_user_get_cookie_store_for_default_user():
assert auth_user.get_cookie_store() == expected_data
def test_auth_not_allowed_for_ghost_user(baseapp):
# NOTE: Importing baseapp is important—if this test runs in isolation,
# the User class may not be bound to the Session, resulting in an error.
# Importing baseapp ensures the session is explicitly bound for this test.
# When tests are run in batch, this binding often happens implicitly as a side effect of other tests or fixtures.
ghost = User.get_ghost_user()
try:
# Under normal conditions, the ghost user should not have a password and should remain inactive.
ghost.password = "<PASSWORD>"
ghost.active = True
Session().commit()
auth_user = auth.AuthUser(user_id=ghost.user_id)
expected_data = {
"username": User.GHOST_USER,
"user_id": ghost.user_id,
"password": md5_safe(ghost.password),
"is_authenticated": False,
}
assert auth_user.get_cookie_store() == expected_data
finally:
# Important: Clean up after this test since the ghost user is unique in the system.
# Not cleaning up could affect other tests or parts of the application.
ghost.password = None
ghost.active = False
Session().commit()
def get_permissions(user, **kwargs):
"""
Utility filling in useful defaults into the call to `_cached_perms_data`.

17
rhodecode/tests/models/test_permissions.py
View file

@ -76,16 +76,23 @@ class TestPermissions(object):
RepoGroupModel().delete(self.g2.group_id)
Session().commit()
UserModel().delete(self.u1, handle_repos="delete", handle_repo_groups="delete")
UserModel().delete(self.u2, handle_repos="delete", handle_repo_groups="delete")
UserModel().delete(self.u3, handle_repos="delete", handle_repo_groups="delete")
UserModel().delete(self.a1, handle_repos="delete", handle_repo_groups="delete")
Session().commit()
self._clean_repos_and_repo_groups()
if hasattr(self, "ug1"):
UserGroupModel().delete(self.ug1, force=True)
Session().commit()
def _clean_repos_and_repo_groups(self):
for u in [self.u1, self.u2, self.u3, self.a1]:
for repo in u.repositories:
RepoModel().delete(repo, forks="detach")
for repo_group in u.repository_groups:
RepoGroupModel().delete(repo_group)
UserModel().delete(u)
Session().commit()
def test_default_perms_set(self, repo_name):
assert repo_perms(self.u1)[repo_name] == "repository.read"
new_perm = "repository.write"