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

Merge pull request !2949 from rhodecode-enterprise-ce security-audit-improvements

Browse source 
Changes from branch: Security audit improvements
This commit is contained in:
Andrii Verbytskyi 2026-01-19 12:53:42 +00:00
commit 01e582003c
29 changed files with 423 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

18
rhodecode/model/db.py
View file

@ -138,7 +138,7 @@ def display_user_sort(obj):
of all other resources
"""
if obj.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if obj.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
return "#####"
prefix = PERMISSION_TYPE_SORT.get(obj.permission.split(".")[-1], "")
extra_sort_num = "1" # default
@ -622,8 +622,10 @@ class User(Base, BaseModel):
DEFAULT_USER_EMAIL = "anonymous@rhodecode.org"
GHOST_USER = "ghost"
AI_USER = "ai_rhodecode"
SECURITY_USER = "security_rhodecode"
GHOST_USER_EMAIL = "ghost@rhodecode.org"
AI_USER_EMAIL = "ai@rhodecode.org"
SECURITY_USER_EMAIL = "security@rhodecode.org"
DEFAULT_GRAVATAR_URL = "https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}"
RECOVERY_CODES_COUNT = 10
@ -732,7 +734,7 @@ class User(Base, BaseModel):
last_activity_long_ago_or_no_activity_since_creation = (cls.last_activity < cutoff_date) | (
(cls.last_activity == null()) & (cls.created_on < cutoff_date)
)
exclude_system_users = ~cls.username.in_([cls.GHOST_USER, cls.AI_USER, cls.DEFAULT_USER])
exclude_system_users = ~cls.username.in_([cls.GHOST_USER, cls.AI_USER, cls.DEFAULT_USER, cls.SECURITY_USER])
query = query.filter(exclude_system_users, last_activity_long_ago_or_no_activity_since_creation)
@ -1323,6 +1325,10 @@ class User(Base, BaseModel):
def get_ai_user(cls, cache=False, refresh=False):
return cls._get_system_user(username=cls.AI_USER, cache=cache, refresh=refresh)
@classmethod
def get_security_user(cls, cache=False, refresh=False):
return cls._get_system_user(username=cls.SECURITY_USER, cache=cache, refresh=refresh)
@classmethod
def get_default_user(cls, cache=False, refresh=False):
return cls._get_system_user(username=cls.DEFAULT_USER, cache=cache, refresh=refresh)
@ -3084,7 +3090,7 @@ class RepoGroup(Base, BaseModel):
@classmethod
def get_user_personal_repo_group(cls, user_id):
user = User.get(user_id)
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
return None
return cls.query().filter(cls.personal == true()).filter(cls.user == user).order_by(cls.group_id.asc()).first()
@ -4516,6 +4522,12 @@ class _PullRequestBase(BaseModel):
default=dict,
)
security_audit_state = Column(
"security_audit_state",
MutationObj.as_mutable(JsonType(dialect_map=dict(mysql=UnicodeText(DEFAULT_JSON_OBJ_SIZE)))),
default=dict,
)
@property
def reviewer_data_json(self):
return str_json(self.reviewer_data)

2
rhodecode/model/forms.py
View file

@ -443,6 +443,7 @@ class _BaseVcsSettingsForm(formencode.Schema):
rhodecode_pr_merge_enabled = v.StringBoolean(if_missing=False)
rhodecode_auto_merge_enabled = v.StringBoolean(if_missing=False)
rhodecode_use_outdated_comments = v.StringBoolean(if_missing=False)
rhodecode_pr_security_scan_enabled = v.StringBoolean(if_missing=False)
# hg
extensions_largefiles = v.StringBoolean(if_missing=False)
@ -671,6 +672,7 @@ def PullRequestForm(localizer, repo_id):
pullrequest_desc = v.UnicodeString(strip=True, required=False)
description_renderer = v.UnicodeString(strip=True, required=False)
close_branch_before_merging = v.StringBoolean(if_missing=False)
run_security_audit = v.StringBoolean(if_missing=False)
return _PullRequestForm

6
rhodecode/model/notification.py
View file

@ -315,6 +315,10 @@ class NotificationModel(BaseModel):
_("%(user)s commented on pull request %(date_or_age)s"),
_("%(user)s commented on pull request at %(date_or_age)s"),
],
EmailNotificationModel.TYPE_SECURITY_AUDIT: [
_("%(user)s completed security audit %(date_or_age)s"),
_("%(user)s completed security audit at %(date_or_age)s"),
],
}
templates = _map[notification.type_]
@ -380,6 +384,7 @@ class EmailNotificationModel(BaseModel):
TYPE_EMAIL_EXCEPTION = "exception"
TYPE_UPDATE_AVAILABLE = "update_available"
TYPE_TEST = "test"
TYPE_SECURITY_AUDIT = "security_audit"
email_types = {
TYPE_MAIN: "rhodecode:templates/email_templates/main.mako",
@ -394,6 +399,7 @@ class EmailNotificationModel(BaseModel):
TYPE_PULL_REQUEST: "rhodecode:templates/email_templates/pull_request_review.mako",
TYPE_PULL_REQUEST_COMMENT: "rhodecode:templates/email_templates/pull_request_comment.mako",
TYPE_PULL_REQUEST_UPDATE: "rhodecode:templates/email_templates/pull_request_update.mako",
TYPE_SECURITY_AUDIT: "rhodecode:templates/email_templates/security_audit.mako",
}
premailer_instance = premailer.Premailer(

7
rhodecode/model/quota_model.py
View file

@ -35,7 +35,12 @@ class UserQuotaModel:
mb_allowance = self.max_disk_space_mb_allowance
repo_count_allowance = self.max_repository_count_allowance
if self.user.admin or self.user.username in [User.AI_USER, User.DEFAULT_USER, User.GHOST_USER]:
if self.user.admin or self.user.username in [
User.AI_USER,
User.DEFAULT_USER,
User.GHOST_USER,
User.SECURITY_USER,
]:
mb_allowance = self.UNLIMITED
repo_count_allowance = self.UNLIMITED

2
rhodecode/model/repo.py
View file

@ -751,7 +751,7 @@ class RepoModel(BaseModel):
if member_type == "user":
member_name = User.get(member_id).username
if member_name in [User.GHOST_USER, User.AI_USER]:
if member_name in [User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise ValueError("Modify user permissions for %s user is not allowed." % member_name)
if member_name == User.DEFAULT_USER:

11
rhodecode/model/repo_group.py
View file

@ -359,7 +359,12 @@ 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 in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if _obj.private and _user_obj.username in [
User.DEFAULT_USER,
User.GHOST_USER,
User.AI_USER,
User.SECURITY_USER,
]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
@ -381,7 +386,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 for default user, just update
if _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if _user_obj.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
log.debug("Skipping private repo %s for user %s", _obj, _user_obj)
return
RepoModel().revoke_user_permission(repo=_obj, user=_user_obj)
@ -426,7 +431,7 @@ class RepoGroupModel(BaseModel):
member_obj = User.get(member_id)
member_name = member_obj.username
is_repo_group = isinstance(obj, RepoGroup) and obj == repo_group
if is_repo_group and member_name in [User.GHOST_USER, User.AI_USER]:
if is_repo_group and member_name in [User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise ValueError("Modify user permissions for %s user is not allowed." % member_name)
if is_repo_group and member_name == User.DEFAULT_USER:

1
rhodecode/model/settings.py
View file

@ -456,6 +456,7 @@ class VcsSettingsModel:
"use_outdated_comments",
"pr_merge_enabled",
"auto_merge_enabled",
"pr_security_scan_enabled",
"hg_use_rebase_for_merging",
"hg_close_branch_before_merging",
"hg_merge_strategy_selector",

8
rhodecode/model/user.py
View file

@ -153,7 +153,7 @@ class UserModel(BaseModel):
qry = (
User.query()
.filter(User.active == true())
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]))
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]))
)
if cache:
qry = qry.options(FromCache("sql_cache_short", "get_active_users"))
@ -189,7 +189,7 @@ class UserModel(BaseModel):
def update_user(self, user, skip_attrs=None, **kwargs):
user = self._get_user(user)
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -373,7 +373,7 @@ class UserModel(BaseModel):
edit = True
# we're not allowed to edit default user or system user
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise DefaultUserException(
"You can't edit this user (`%(username)s`) since it's "
"crucial for entire application" % {"username": user.username}
@ -594,7 +594,7 @@ class UserModel(BaseModel):
user = self._get_user(user)
try:
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise DefaultUserException("You can't remove this user since it's crucial for entire application")
if delete_reviewer: