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

Initial implementation of security scan for PRs

Browse source
This commit is contained in:
Andrii V 2026-01-15 16:48:04 +01:00
parent 511c54b606
commit 9ce6ca8fd8
13 changed files with 273 additions and 1 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__ = 121 # defines current db version for migrations
__dbversion__ = 123 # defines current db version for migrations
__license__ = "AGPLv3, and Commercial License"
__author__ = "RhodeCode GmbH"
__url__ = "https://code.rhodecode.com"

62
rhodecode/apps/repository/views/repo_pull_requests.py
View file

@ -450,6 +450,11 @@ class RepoPullRequestsView(RepoAppView, DataGridAppView):
if ai_pr_state:
c.ai_code_review_state = ai_pr_state.get("review_state", None)
# Add security audit state
c.security_audit_state = None
if security_state := pull_request.security_audit_state:
c.security_audit_state = security_state
# inject latest version
latest_ver = PullRequest.get_pr_display_object(pull_request_latest, pull_request_latest)
c.versions = versions + [latest_ver]
@ -1274,6 +1279,35 @@ class RepoPullRequestsView(RepoAppView, DataGridAppView):
auth_user=self._rhodecode_user,
settings=settings,
)
# Handle security audit if requested (EE only)
run_security_audit = _form.get("run_security_audit", False)
if run_security_audit:
try:
from rc_ee.lib.celerylib import tasks as ee_tasks
from rhodecode.lib.celerylib import run_task
import datetime
# Initialize security audit state
pull_request.security_audit_state = {
"status": "pending",
"started_at": datetime.datetime.utcnow().isoformat(),
"started_by": self._rhodecode_user.user_id,
}
# Trigger async security audit task
run_task(
ee_tasks.audit_pull_request_diff, pull_request.pull_request_id, self._rhodecode_user.user_id
)
log.debug(
"Security audit triggered for PR #%s by user %s",
pull_request.pull_request_id,
self._rhodecode_user.username,
)
except ImportError:
log.warning("EE security audit tasks not available, skipping")
Session().commit()
h.flash(_("Successfully opened new pull request"), category="success")
@ -1550,6 +1584,34 @@ class RepoPullRequestsView(RepoAppView, DataGridAppView):
)
h.flash(msg, category="success")
channelstream.pr_update_channelstream_push(self.request, c.pr_broadcast_channel, self._rhodecode_user, msg)
# Re-run security audit if it was enabled for this PR
if pull_request.security_audit_state and pull_request.security_audit_state.get("status"):
try:
from rc_ee.lib.celerylib import tasks as ee_tasks
from rhodecode.lib.celerylib import run_task
import datetime
# Reset state to pending for re-scan
pull_request.security_audit_state = {
"status": "pending",
"started_at": datetime.datetime.utcnow().isoformat(),
"started_by": self._rhodecode_user.user_id,
}
Session().commit()
# Trigger async security audit task
run_task(
ee_tasks.audit_pull_request_diff, pull_request.pull_request_id, self._rhodecode_user.user_id
)
log.debug(
"Security audit re-triggered for PR #%s after update by user %s",
pull_request.pull_request_id,
self._rhodecode_user.username,
)
except ImportError:
log.warning("EE security audit tasks not available, skipping re-scan")
else:
msg = PullRequestModel.UPDATE_STATUS_MESSAGES[resp.reason]
warning_reasons = [

15
rhodecode/lib/db_manage.py
View file

@ -632,6 +632,21 @@ class DbManage(object):
Session().add(ai_user)
Session().commit()
def create_security_user(self):
if not User.get_by_username(User.SECURITY_USER):
log.info("creating security user")
security_user = User()
security_user.name = "Security"
security_user.username = User.SECURITY_USER
security_user.lastname = "Scanner"
security_user.strict_creation_check = False
security_user.email = User.SECURITY_USER_EMAIL
security_user.active = False
security_user.description = "System user for security scanning"
security_user.is_new_user = False
Session().add(security_user)
Session().commit()
def create_default_user(self):
log.info("creating default user")
# create default user for handling default permissions.

48
rhodecode/lib/dbmigrate/versions/122_version_5_10_0.py Normal file
View file

@ -0,0 +1,48 @@
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
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)
context = MigrationContext.configure(migrate_engine.connect())
op = Operations(context)
inspector = inspect(migrate_engine)
pr_tables = [db.PullRequest.__table__, db.PullRequestVersion.__table__]
for pr_table in pr_tables:
existing_columns = [col["name"] for col in inspector.get_columns(pr_table.name)]
new_column_name = "security_audit_state"
if new_column_name not in existing_columns:
with op.batch_alter_table(pr_table.name) as batch_op:
new_column = Column(
new_column_name,
MutationObj.as_mutable(
JsonType(dialect_map=dict(mysql=UnicodeText(16384))),
),
default=dict,
)
batch_op.add_column(new_column)
def downgrade(migrate_engine):
pass

37
rhodecode/lib/dbmigrate/versions/123_version_5_10_0.py Normal file
View file

@ -0,0 +1,37 @@
from rhodecode.lib.dbmigrate.versions import _reset_base
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
username = "security_rhodecode"
if not User.get_by_username(username):
security_user = UserModel()
security_user.name = "Security"
security_user.username = username
security_user.lastname = "Scanner"
security_user.strict_creation_check = False
security_user.email = "security@rhodecode.org"
security_user.active = False
security_user.description = "System user for security scanning"
security_user.is_new_user = False
session = meta.Session()
session.add(security_user)
session.commit()
def downgrade(migrate_engine):
pass

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

@ -98,6 +98,7 @@ def command(
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_ai_user()
dbmanage.create_security_user()
dbmanage.create_admin_and_prompt()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()

1
rhodecode/lib/utils.py
View file

@ -843,6 +843,7 @@ def create_test_database(test_path, config):
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_ai_user()
dbmanage.create_security_user()
dbmanage.create_test_admin_and_users()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()

12
rhodecode/model/db.py
View file

@ -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
@ -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)
@ -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)

1
rhodecode/model/forms.py
View file

@ -671,6 +671,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

2
rhodecode/model/notification.py
View file

@ -380,6 +380,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 +395,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(

28
rhodecode/templates/email_templates/security_audit.mako Normal file
View file

@ -0,0 +1,28 @@
<%inherit file="base.mako"/>
<%def name="subject()" filter="n,trim,whitespace_filter">
Security Audit Completed: ${repo_name}
</%def>
<%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)}
---
${self.plaintext_footer()}
</%def>
<table style="text-align:left;vertical-align:top;">
<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>
</tr>
</table>

18
rhodecode/templates/pullrequests/pullrequest.mako
View file

@ -229,6 +229,24 @@
</label>
</div>
% endif
% if c.rhodecode_edition_id == 'EE':
<div class="pull-request-settings">
${h.checkbox('run_security_audit', checked=False, value=True)}
<label for="run_security_audit">
${h.literal(_('Run security audit on PR diff'))}
<span class="tooltip" title="${_('Scan only the changes in this pull request for exposed secrets and credentials. Results will appear in the PR sidebar.')}">[?]</span>
</label>
</div>
% else:
<div class="pull-request-settings">
${h.checkbox('run_security_audit', checked=False, value=True, disabled=True, **{'class': 'disabled-checkbox'})}
<label for="run_security_audit" class="disabled-label">
${h.literal(_('Run security audit on PR diff'))}
<span class="ee-feature-label">[EE Only]</span>
</label>
</div>
% endif
</div>
</div>
</div>

47
rhodecode/templates/pullrequests/pullrequest_show.mako
View file

@ -731,6 +731,53 @@
</div>
% endif
% if c.rhodecode_edition_id == 'EE' and c.security_audit_state:
<div id="security-audit-section" class="sidebar-element clear-both">
<div class="right-sidebar-expanded-state pr-details-title">
<span class="sidebar-heading">
<i class="icon-lock"></i>
${_('Security Audit')}
</span>
</div>
<div style="margin-top: 10px">
% if c.security_audit_state.get('status') == 'pending':
<span class="tag" style="background-color: #ffc854; color: #333;">
<i class="icon-hourglass"></i>
${_('Audit Pending')}
</span>
% elif c.security_audit_state.get('status') == 'scanning':
<span class="tag" style="background-color: #5bc0de; color: #fff;">
<i class="icon-refresh"></i>
${_('Scanning...')}
</span>
% elif c.security_audit_state.get('status') == 'completed':
<%
count = c.security_audit_state.get('findings_count', 0)
%>
% if count > 0:
<span class="tag" style="background-color: #ffc854; color: #333;">
<i class="icon-attention"></i>
${_('%s finding(s)') % count}
</span>
<p style="font-size: 11px; color: #666; margin-top: 5px;">
${_('See inline comments for details')}
</p>
% else:
<span class="tag" style="background-color: #0ac878; color: #fff;">
<i class="icon-ok"></i>
${_('No secrets found')}
</span>
% endif
% elif c.security_audit_state.get('status') == 'failed':
<span class="tag" style="background-color: #e85e4d; color: #fff;">
<i class="icon-attention"></i>
${_('Audit Failed')}
</span>
% endif
</div>
</div>
% endif
## TODOs
<div id="todosTable" class="sidebar-element clear-both">
<div class="tooltip right-sidebar-collapsed-state" style="display: none" onclick="toggleSidebar(); return false" title="TODOs">