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

New implementations and updates for security scan

Browse source
This commit is contained in:
Andrii V 2026-01-16 16:53:31 +01:00
parent 9ce6ca8fd8
commit 6fe5736648
18 changed files with 145 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -33,7 +33,7 @@ class TestGetUsers(object):
ret_all = []
_users = (
User.query()
.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]))
.order_by(User.username)
.all()
)

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

@ -148,7 +148,7 @@ def get_users(request, apiuser):
users_list = (
User.query()
.order_by(User.username)
.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]))
.all()
)
for user in users_list:

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

@ -168,7 +168,7 @@ class BaseAppView(object):
if not user_obj:
return
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]:
return
now = time.time()
@ -294,7 +294,7 @@ class RepoAppView(BaseAppView):
def _prepare_and_set_clone_url(self, c):
username = ""
if self._rhodecode_user.username not in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if self._rhodecode_user.username not in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
username = self._rhodecode_user.username
_def_clone_uri = c.clone_uri_tmpl
@ -591,7 +591,7 @@ class UserAppView(BaseAppView):
_ = self.request.translate
if not request.db_user_supports_default:
if self.db_user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if self.db_user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
h.flash(
_("Editing user `{}` is disabled.".format(self.db_user.username)),
category="warning",

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

@ -42,7 +42,9 @@ class TestAdminUsersView(TestController):
response = self.app.get(route_path("users_data"), extra_environ=xhr_header)
all_users = (
User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER])).count()
User.query()
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]))
.count()
)
assert response.json["recordsTotal"] == all_users
@ -53,7 +55,9 @@ class TestAdminUsersView(TestController):
)
all_users = (
User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER])).count()
User.query()
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]))
.count()
)
assert response.json["recordsTotal"] == all_users
assert response.json["recordsFiltered"] == 0

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

@ -106,7 +106,7 @@ class AdminUsersView(BaseAppView, DataGridAppView):
def user_actions(user_id, username):
return _render("user_actions", user_id, username)
excluded_users = User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER])
excluded_users = User.username.notin_([User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER])
users_data_total_count = User.query().filter(excluded_users).count()
users_data_total_inactive_count = User.query().filter(excluded_users).filter(User.active != true()).count()

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

@ -244,7 +244,7 @@ class HomeView(BaseAppView, DataGridAppView):
User.query()
.order_by(func.length(User.username))
.order_by(User.username)
.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 name_contains:

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 in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]:
if not c.user or c.user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]:
raise HTTPNotFound()
return self._get_template_context(c)

11
rhodecode/lib/celerylib/tasks.py
View file

@ -124,6 +124,15 @@ def send_email(recipients, subject, body="", html_body="", email_config=None, ex
try:
mailer = get_mailer(transformed_email_confing, email_config)
log.debug(
"Preparing email: server=%s, sender=%s, recipients=%s, subject=%s, dev_mode=%s",
mail_server,
transformed_email_confing.get("default_sender"),
recipients,
subject,
dev_mod,
)
message = Message(
subject=subject,
sender=transformed_email_confing["default_sender"],
@ -132,7 +141,9 @@ def send_email(recipients, subject, body="", html_body="", email_config=None, ex
html=html_body,
extra_headers=extra_headers,
)
log.debug("Sending email via mailer...")
mailer.send_immediately(message)
log.info("Email sent successfully to %s", recipients)
statsd = StatsdClient.statsd
if statsd:

2
rhodecode/lib/helpers.py
View file

@ -1036,7 +1036,7 @@ def link_to_user(author, length=0, **kwargs):
if length:
display_person = shorter(display_person, length)
if user and user.username not in [user.DEFAULT_USER, user.GHOST_USER, user.AI_USER]:
if user and user.username not in [user.DEFAULT_USER, user.GHOST_USER, user.AI_USER, user.SECURITY_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 not in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER]
and user_obj.username not in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]
and user_obj.user_data.get("force_password_change")
):
reason = "password change required"

5
rhodecode/lib/utils.py
View file

@ -879,7 +879,10 @@ def create_test_repositories(test_path, config):
def password_changed(auth_user, session):
# Never report password change in case of default user, anonymous user or system user.
if auth_user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER] or auth_user.user_id is None:
if (
auth_user.username in [User.DEFAULT_USER, User.GHOST_USER, User.AI_USER, User.SECURITY_USER]
or auth_user.user_id is None
):
return False
password_hash = md5(safe_bytes(auth_user.password)) if auth_user.password else None

6
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
@ -734,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)
@ -3090,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()

4
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_]

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:

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:

96
rhodecode/templates/email_templates/security_audit.mako
View file

@ -1,28 +1,108 @@
<%inherit file="base.mako"/>
<%namespace name="base" file="base.mako"/>
## EMAIL SUBJECT
<%def name="subject()" filter="n,trim,whitespace_filter">
Security Audit Completed: ${repo_name}
</%def>
## PLAINTEXT VERSION OF BODY
<%def name="body_plaintext()" filter="n,trim">
Security Audit Results for ${repo_name}
${body}
Files Scanned: ${scan_summary.get('files_scanned', 0)}
Secrets Found: ${scan_summary.get('secrets_found', 0)}
% if scan_summary.get('secrets_found', 0) > 0:
FINDINGS:
% for finding in findings:
- ${finding['file']} (line ${finding['line']}): ${finding['type']}
% endfor
% else:
No exposed credentials or secrets were found.
% endif
---
${self.plaintext_footer()}
</%def>
<table style="text-align:left;vertical-align:top;">
## header
<table style="text-align:left;vertical-align:middle;width: 100%">
<tr>
<td style="padding:20px;">
<h3>Security Audit Results for ${repo_name}</h3>
<div style="white-space:pre-wrap">${body}</div>
<p><strong>Files Scanned:</strong> ${scan_summary.get('files_scanned', 0)}</p>
<p><strong>Secrets Found:</strong> ${scan_summary.get('secrets_found', 0)}</p>
<td style="width:100%;border-bottom:1px solid #dbd9da;">
<div style="margin: 0; font-weight: bold">
<div class="clear-both" style="margin-bottom: 4px">
<span style="color:#7E7F7F">${_('Security Scanner')}</span>
${_('completed audit for repository')}
<a href="${repo_url}" style="${base.link_css()}">${repo_name}</a>
</div>
</div>
</td>
</tr>
</table>
<div class="clear-both"></div>
## main body
<table style="text-align:left;vertical-align:middle;width: 100%">
<tr>
<td style="width: 130px"></td>
<td></td>
</tr>
<tr>
<td style="padding-right:20px;"><strong>${_('Repository')}:</strong></td>
<td>
<a href="${repo_url}" style="${base.link_css()}">${repo_name}</a>
</td>
</tr>
<tr>
<td style="padding-right:20px;"><strong>${_('Files Scanned')}:</strong></td>
<td>${scan_summary.get('files_scanned', 0)}</td>
</tr>
<tr>
<td style="padding-right:20px;"><strong>${_('Secrets Found')}:</strong></td>
<td>
% if scan_summary.get('secrets_found', 0) > 0:
<span style="color: #e85e4d; font-weight: bold;">${scan_summary.get('secrets_found', 0)}</span>
% else:
<span style="color: #0ac878; font-weight: bold;">0</span>
% endif
</td>
</tr>
</table>
% if scan_summary.get('secrets_found', 0) > 0:
<div style="margin-top: 20px;">
<table style="text-align:left;vertical-align:middle;width: 100%; border-collapse: collapse;">
<tr style="background-color: #f5f5f5;">
<th style="padding: 8px; border: 1px solid #ddd; text-align: left;">${_('File')}</th>
<th style="padding: 8px; border: 1px solid #ddd; text-align: left;">${_('Line')}</th>
<th style="padding: 8px; border: 1px solid #ddd; text-align: left;">${_('Type')}</th>
</tr>
% for finding in findings:
<tr>
<td style="padding: 8px; border: 1px solid #ddd;"><code>${finding['file']}</code></td>
<td style="padding: 8px; border: 1px solid #ddd;">${finding['line']}</td>
<td style="padding: 8px; border: 1px solid #ddd;">${finding['type']}</td>
</tr>
% endfor
</table>
</div>
<div style="margin-top: 15px; padding: 10px; background-color: #fff3cd; border: 1px solid #ffc107; border-radius: 4px;">
<strong style="color: #856404;">${_('Action Required')}</strong>
<p style="margin: 5px 0 0 0; color: #856404;">
${_('Please review the findings above and remove or rotate any exposed credentials.')}
</p>
</div>
% else:
<div style="margin-top: 15px; padding: 10px; background-color: #d4edda; border: 1px solid #28a745; border-radius: 4px;">
<strong style="color: #155724;">${_('All Clear')}</strong>
<p style="margin: 5px 0 0 0; color: #155724;">
${_('No exposed credentials or secrets were found in this repository.')}
</p>
</div>
% endif