# Copyright (C) 2011-2024 RhodeCode GmbH # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License, version 3 # (only), as published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . # # This program is dual-licensed. If you wish to learn more about the # RhodeCode Enterprise Edition, including its added features, Support services, # and proprietary license terms, please see https://rhodecode.com/licenses/ import itertools import logging import os import collections import urllib.request import urllib.parse import urllib.error import pathlib import time import random from pyramid.httpexceptions import HTTPNotFound, HTTPBadRequest, HTTPFound from pyramid.renderers import render from pyramid.response import Response import rhodecode from rhodecode.apps._base import RepoAppView from rhodecode.lib import diffs, helpers as h, rc_cache from rhodecode.lib import audit_logger from rhodecode.lib.hash_utils import sha1_safe from rhodecode.lib.archive_cache import ( get_archival_cache_store, get_archival_config, ArchiveCacheGenerationLock, archive_iterator) from rhodecode.lib.str_utils import safe_bytes, convert_special_chars from rhodecode.lib.view_utils import parse_path_ref from rhodecode.lib.exceptions import NonRelativePathError from rhodecode.lib.codeblocks import ( filenode_as_lines_tokens, filenode_as_annotated_lines_tokens) from rhodecode.lib.utils2 import convert_line_endings, detect_mode from rhodecode.lib.type_utils import str2bool from rhodecode.lib.str_utils import safe_str, safe_int, header_safe_str from rhodecode.lib.auth import ( LoginRequired, HasRepoPermissionAnyDecorator, CSRFRequired) from rhodecode.lib.vcs import path as vcspath from rhodecode.lib.vcs.backends.base import EmptyCommit from rhodecode.lib.vcs.conf import settings from rhodecode.lib.vcs.nodes import FileNode from rhodecode.lib.vcs.exceptions import ( RepositoryError, CommitDoesNotExistError, EmptyRepositoryError, ImproperArchiveTypeError, VCSError, NodeAlreadyExistsError, NodeDoesNotExistError, CommitError, NodeError) from rhodecode.model.scm import ScmModel from rhodecode.model.db import Repository log = logging.getLogger(__name__) def get_archive_name(db_repo_id, db_repo_name, commit_sha, ext, subrepos=False, path_sha='', with_hash=True): # original backward compat name of archive clean_name = safe_str(convert_special_chars(db_repo_name).replace('/', '_')) # e.g vcsserver-id-abcd-sub-1-abcfdef-archive-all.zip # vcsserver-id-abcd-sub-0-abcfdef-COMMIT_SHA-PATH_SHA.zip id_sha = sha1_safe(str(db_repo_id))[:4] sub_repo = 'sub-1' if subrepos else 'sub-0' commit = commit_sha if with_hash else 'archive' path_marker = (path_sha if with_hash else '') or 'all' archive_name = f'{clean_name}-id-{id_sha}-{sub_repo}-{commit}-{path_marker}{ext}' return archive_name def get_path_sha(at_path): return safe_str(sha1_safe(at_path)[:8]) def _get_archive_spec(fname): log.debug('Detecting archive spec for: `%s`', fname) fileformat = None ext = None content_type = None for a_type, content_type, extension in settings.ARCHIVE_SPECS: if fname.endswith(extension): fileformat = a_type log.debug('archive is of type: %s', fileformat) ext = extension break if not fileformat: raise ValueError() # left over part of whole fname is the commit commit_id = fname[:-len(ext)] return commit_id, ext, fileformat, content_type class RepoFilesView(RepoAppView): @staticmethod def adjust_file_path_for_svn(f_path, repo): """ Computes the relative path of `f_path`. This is mainly based on prefix matching of the recognized tags and branches in the underlying repository. """ tags_and_branches = itertools.chain( repo.branches.keys(), repo.tags.keys()) tags_and_branches = sorted(tags_and_branches, key=len, reverse=True) for name in tags_and_branches: if f_path.startswith(f'{name}/'): f_path = vcspath.relpath(f_path, name) break return f_path def load_default_context(self): c = self._get_local_tmpl_context(include_app_defaults=True) c.rhodecode_repo = self.rhodecode_vcs_repo c.enable_downloads = self.db_repo.enable_downloads return c def _ensure_not_locked(self, commit_id='tip'): _ = self.request.translate repo = self.db_repo if repo.enable_locking and repo.locked[0]: h.flash(_('This repository has been locked by %s on %s') % (h.person_by_id(repo.locked[0]), h.format_date(h.time_to_datetime(repo.locked[1]))), 'warning') files_url = h.route_path( 'repo_files:default_path', repo_name=self.db_repo_name, commit_id=commit_id) raise HTTPFound(files_url) def forbid_non_head(self, is_head, f_path, commit_id='tip', json_mode=False): _ = self.request.translate if not is_head: message = _('Cannot modify file. ' 'Given commit `{}` is not head of a branch.').format(commit_id) h.flash(message, category='warning') if json_mode: return message files_url = h.route_path( 'repo_files', repo_name=self.db_repo_name, commit_id=commit_id, f_path=f_path) raise HTTPFound(files_url) def check_branch_permission(self, branch_name, commit_id='tip', json_mode=False): _ = self.request.translate rule, branch_perm = self._rhodecode_user.get_rule_and_branch_permission( self.db_repo_name, branch_name) if branch_perm and branch_perm not in ['branch.push', 'branch.push_force']: message = _('Branch `{}` changes forbidden by rule {}.').format( h.escape(branch_name), h.escape(rule)) h.flash(message, 'warning') if json_mode: return message files_url = h.route_path( 'repo_files:default_path', repo_name=self.db_repo_name, commit_id=commit_id) raise HTTPFound(files_url) def _get_commit_and_path(self): default_commit_id = self.db_repo.landing_ref_name default_f_path = '/' commit_id = self.request.matchdict.get('commit_id', default_commit_id) f_path = self._get_f_path(self.request.matchdict, default_f_path) bytes_path = safe_bytes(f_path) return commit_id, f_path, bytes_path @classmethod def _get_default_encoding(cls, c): enc_list = getattr(c, 'default_encodings', []) return enc_list[0] if enc_list else 'UTF-8' def _get_commit_or_redirect(self, commit_id, redirect_after=True): """ This is a safe way to get commit. If an error occurs it redirects to tip with proper message :param commit_id: id of commit to fetch :param redirect_after: toggle redirection """ _ = self.request.translate try: return self.rhodecode_vcs_repo.get_commit(commit_id) except EmptyRepositoryError: if not redirect_after: return None add_new = upload_new = "" if h.HasRepoPermissionAny( 'repository.write', 'repository.admin')(self.db_repo_name): _url = h.route_path( 'repo_files_add_file', repo_name=self.db_repo_name, commit_id=0, f_path='') add_new = h.link_to( _('add a new file'), _url, class_="alert-link") _url_upld = h.route_path( 'repo_files_upload_file', repo_name=self.db_repo_name, commit_id=0, f_path='') upload_new = h.link_to( _('upload a new file'), _url_upld, class_="alert-link") h.flash(h.literal( _('There are no files yet. Click here to %s or %s.') % (add_new, upload_new)), category='warning') raise HTTPFound( h.route_path('repo_summary', repo_name=self.db_repo_name)) except (CommitDoesNotExistError, LookupError) as e: msg = _('No such commit exists for this repository. Commit: {}').format(commit_id) h.flash(msg, category='error') raise HTTPNotFound() except RepositoryError as e: h.flash(h.escape(safe_str(e)), category='error') raise HTTPNotFound() def _get_filenode_or_redirect(self, commit_obj, path, pre_load=None): """ Returns file_node, if error occurs or given path is directory, it'll redirect to top level path """ _ = self.request.translate try: file_node = commit_obj.get_node(path, pre_load=pre_load) if file_node.is_dir(): raise RepositoryError('The given path is a directory') except CommitDoesNotExistError: log.exception('No such commit exists for this repository') h.flash(_('No such commit exists for this repository'), category='error') raise HTTPNotFound() except RepositoryError as e: log.warning('Repository error while fetching filenode `%s`. Err:%s', path, e) h.flash(h.escape(safe_str(e)), category='error') raise HTTPNotFound() return file_node def _is_valid_head(self, commit_id, repo, landing_ref): branch_name = sha_commit_id = '' is_head = False log.debug('Checking if commit_id `%s` is a head for %s.', commit_id, repo) for _branch_name, branch_commit_id in repo.branches.items(): # simple case we pass in branch name, it's a HEAD if commit_id == _branch_name: is_head = True branch_name = _branch_name sha_commit_id = branch_commit_id break # case when we pass in full sha commit_id, which is a head elif commit_id == branch_commit_id: is_head = True branch_name = _branch_name sha_commit_id = branch_commit_id break if h.is_svn(repo) and not repo.is_empty(): # Note: Subversion only has one head. if commit_id == repo.get_commit(commit_idx=-1).raw_id: is_head = True return branch_name, sha_commit_id, is_head # checked branches, means we only need to try to get the branch/commit_sha if repo.is_empty(): is_head = True branch_name = landing_ref sha_commit_id = EmptyCommit().raw_id else: commit = repo.get_commit(commit_id=commit_id) if commit: branch_name = commit.branch sha_commit_id = commit.raw_id return branch_name, sha_commit_id, is_head def _get_tree_at_commit(self, c, commit_id, f_path, full_load=False, at_rev=None): repo_id = self.db_repo.repo_id force_recache = self.get_recache_flag() cache_seconds = safe_int( rhodecode.CONFIG.get('rc_cache.cache_repo.expiration_time')) cache_on = not force_recache and cache_seconds > 0 log.debug( 'Computing FILE TREE for repo_id %s commit_id `%s` and path `%s`' 'with caching: %s[TTL: %ss]' % ( repo_id, commit_id, f_path, cache_on, cache_seconds or 0)) cache_namespace_uid = f'repo.{rc_cache.FILE_TREE_CACHE_VER}.{repo_id}' region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid) @region.conditional_cache_on_arguments(namespace=cache_namespace_uid, condition=cache_on) def compute_file_tree(_name_hash, _repo_id, _commit_id, _f_path, _full_load, _at_rev): log.debug('Generating cached file tree at for repo_id: %s, %s, %s', _repo_id, _commit_id, _f_path) c.full_load = _full_load return render( 'rhodecode:templates/files/files_browser_tree.mako', self._get_template_context(c), self.request, _at_rev) return compute_file_tree( self.db_repo.repo_name_hash, self.db_repo.repo_id, commit_id, f_path, full_load, at_rev) def create_pure_path(self, *parts): # Split paths and sanitize them, removing any ../ etc sanitized_path = [ x for x in pathlib.PurePath(*parts).parts if x not in ['.', '..']] pure_path = pathlib.PurePath(*sanitized_path) return pure_path def _is_lf_enabled(self, target_repo): lf_enabled = False lf_key_for_vcs_map = { 'hg': 'extensions_largefiles', 'git': 'vcs_git_lfs_enabled' } lf_key_for_vcs = lf_key_for_vcs_map.get(target_repo.repo_type) if lf_key_for_vcs: lf_enabled = self._get_repo_setting(target_repo, lf_key_for_vcs) return lf_enabled @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_archivefile(self): # archive cache config from rhodecode import CONFIG _ = self.request.translate self.load_default_context() subrepos = self.request.GET.get('subrepos') == 'true' with_hash = str2bool(self.request.GET.get('with_hash', '1')) default_at_path = '/' fname = self.request.matchdict['fname'] at_path = self.request.GET.get('at_path') or default_at_path if not self.db_repo.enable_downloads: return Response(_('Downloads disabled')) try: commit_id, ext, file_format, content_type = _get_archive_spec(fname) except ValueError: return Response(_('Unknown archive type for: `{}`').format(h.escape(fname))) try: commit = self.rhodecode_vcs_repo.get_commit(commit_id) except CommitDoesNotExistError: return Response(_('Unknown commit_id {}').format( h.escape(commit_id))) except EmptyRepositoryError: return Response(_('Empty repository')) # we used a ref, or a shorter version, lets redirect client ot use explicit hash if commit_id != commit.raw_id: fname=f'{commit.raw_id}{ext}' raise HTTPFound(self.request.current_route_path(fname=fname)) try: at_path = commit.get_node(safe_bytes(at_path)).path or default_at_path except Exception: return Response(_('No node at path {} for this repository').format(h.escape(at_path))) path_sha = get_path_sha(at_path) # used for cache etc, consistent unique archive name archive_name_key = get_archive_name( self.db_repo.repo_id, self.db_repo_name, commit_sha=commit.short_id, ext=ext, subrepos=subrepos, path_sha=path_sha, with_hash=True) if not with_hash: path_sha = '' # what end client gets served response_archive_name = get_archive_name( self.db_repo.repo_id, self.db_repo_name, commit_sha=commit.short_id, ext=ext, subrepos=subrepos, path_sha=path_sha, with_hash=with_hash) # remove extension from our archive directory name archive_dir_name = response_archive_name[:-len(ext)] archive_cache_disable = self.request.GET.get('no_cache') d_cache = get_archival_cache_store(config=CONFIG) # NOTE: we get the config to pass to a call to lazy-init the SAME type of cache on vcsserver d_cache_conf = get_archival_config(config=CONFIG) # This is also a cache key, and lock key reentrant_lock_key = archive_name_key + '.lock' use_cached_archive = False if not archive_cache_disable and archive_name_key in d_cache: reader, metadata = d_cache.fetch(archive_name_key) use_cached_archive = True log.debug('Found cached archive as key=%s tag=%s, serving archive from cache reader=%s', archive_name_key, metadata, reader.name) else: reader = None log.debug('Archive with key=%s is not yet cached, creating one now...', archive_name_key) if not reader: # generate new archive, as previous was not found in the cache try: with d_cache.get_lock(reentrant_lock_key): try: commit.archive_repo(archive_name_key, archive_dir_name=archive_dir_name, kind=file_format, subrepos=subrepos, archive_at_path=at_path, cache_config=d_cache_conf) except ImproperArchiveTypeError: return _('Unknown archive type') except ArchiveCacheGenerationLock: retry_after = round(random.uniform(0.3, 3.0), 1) time.sleep(retry_after) location = self.request.url response = Response( f"archive {archive_name_key} generation in progress, Retry-After={retry_after}, Location={location}" ) response.headers["Retry-After"] = str(retry_after) response.status_code = 307 # temporary redirect response.location = location return response reader, metadata = d_cache.fetch(archive_name_key, retry=True, retry_attempts=30) response = Response(app_iter=archive_iterator(reader)) response.content_disposition = f'attachment; filename={response_archive_name}' response.content_type = str(content_type) try: return response finally: # store download action audit_logger.store_web( 'repo.archive.download', action_data={ 'user_agent': self.request.user_agent, 'archive_name': archive_name_key, 'archive_spec': fname, 'archive_cached': use_cached_archive}, user=self._rhodecode_user, repo=self.db_repo, commit=True ) def _get_file_node(self, commit_id, f_path): if commit_id not in ['', None, 'None', '0' * 12, '0' * 40]: commit = self.rhodecode_vcs_repo.get_commit(commit_id=commit_id) try: node = commit.get_node(safe_bytes(f_path)) if node.is_dir(): raise NodeError(f'{node} path is a {type(node)} not a file') except NodeDoesNotExistError: commit = EmptyCommit( commit_id=commit_id, idx=commit.idx, repo=commit.repository, alias=commit.repository.alias, message=commit.message, author=commit.author, date=commit.date) node = FileNode(safe_bytes(f_path), b'', commit=commit) else: commit = EmptyCommit( repo=self.rhodecode_vcs_repo, alias=self.rhodecode_vcs_repo.alias) node = FileNode(safe_bytes(f_path), b'', commit=commit) return node @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_files_diff(self): c = self.load_default_context() f_path = self._get_f_path(self.request.matchdict) diff1 = self.request.GET.get('diff1', '') diff2 = self.request.GET.get('diff2', '') path1, diff1 = parse_path_ref(diff1, default_path=f_path) ignore_whitespace = str2bool(self.request.GET.get('ignorews')) line_context = self.request.GET.get('context', 3) if not any((diff1, diff2)): h.flash( 'Need query parameter "diff1" or "diff2" to generate a diff.', category='error') raise HTTPBadRequest() c.action = self.request.GET.get('diff') if c.action not in ['download', 'raw']: compare_url = h.route_path( 'repo_compare', repo_name=self.db_repo_name, source_ref_type='rev', source_ref=diff1, target_repo=self.db_repo_name, target_ref_type='rev', target_ref=diff2, _query=dict(f_path=f_path)) # redirect to new view if we render diff raise HTTPFound(compare_url) try: node1 = self._get_file_node(diff1, path1) node2 = self._get_file_node(diff2, f_path) except (RepositoryError, NodeError): log.exception("Exception while trying to get node from repository") raise HTTPFound( h.route_path('repo_files', repo_name=self.db_repo_name, commit_id='tip', f_path=f_path)) if all(isinstance(node.commit, EmptyCommit) for node in (node1, node2)): raise HTTPNotFound() c.commit_1 = node1.commit c.commit_2 = node2.commit if c.action == 'download': _diff = diffs.get_gitdiff(node1, node2, ignore_whitespace=ignore_whitespace, context=line_context) # NOTE: this was using diff_format='gitdiff' diff = diffs.DiffProcessor(_diff, diff_format='newdiff') response = Response(self.path_filter.get_raw_patch(diff)) response.content_type = 'text/plain' response.content_disposition = ( f'attachment; filename={f_path}_{diff1}_vs_{diff2}.diff' ) charset = self._get_default_encoding(c) if charset: response.charset = charset return response elif c.action == 'raw': _diff = diffs.get_gitdiff(node1, node2, ignore_whitespace=ignore_whitespace, context=line_context) # NOTE: this was using diff_format='gitdiff' diff = diffs.DiffProcessor(_diff, diff_format='newdiff') response = Response(self.path_filter.get_raw_patch(diff)) response.content_type = 'text/plain' charset = self._get_default_encoding(c) if charset: response.charset = charset return response # in case we ever end up here raise HTTPNotFound() @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_files_diff_2way_redirect(self): """ Kept only to make OLD links work """ f_path = self._get_f_path_unchecked(self.request.matchdict) diff1 = self.request.GET.get('diff1', '') diff2 = self.request.GET.get('diff2', '') if not any((diff1, diff2)): h.flash( 'Need query parameter "diff1" or "diff2" to generate a diff.', category='error') raise HTTPBadRequest() compare_url = h.route_path( 'repo_compare', repo_name=self.db_repo_name, source_ref_type='rev', source_ref=diff1, target_ref_type='rev', target_ref=diff2, _query=dict(f_path=f_path, diffmode='sideside', target_repo=self.db_repo_name,)) raise HTTPFound(compare_url) @LoginRequired() def repo_files_default_commit_redirect(self): """ Special page that redirects to the landing page of files based on the default commit for repository """ c = self.load_default_context() ref_name = c.rhodecode_db_repo.landing_ref_name landing_url = h.repo_files_by_ref_url( c.rhodecode_db_repo.repo_name, c.rhodecode_db_repo.repo_type, f_path='', ref_name=ref_name, commit_id='tip', query=dict(at=ref_name) ) raise HTTPFound(landing_url) @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_files(self): c = self.load_default_context() view_name = getattr(self.request.matched_route, 'name', None) c.annotate = view_name == 'repo_files:annotated' # default is false, but .rst/.md files later are auto rendered, we can # overwrite auto rendering by setting this GET flag c.renderer = view_name == 'repo_files:rendered' or not self.request.GET.get('no-render', False) commit_id, f_path, bytes_path = self._get_commit_and_path() c.commit = self._get_commit_or_redirect(commit_id) c.branch = self.request.GET.get('branch', None) c.f_path = f_path at_rev = self.request.GET.get('at') # files or dirs try: c.file = c.commit.get_node(bytes_path, pre_load=['is_binary', 'size', 'data']) c.file_author = True c.file_tree = '' # prev link try: prev_commit = c.commit.prev(c.branch) c.prev_commit = prev_commit c.url_prev = h.route_path('repo_files', repo_name=self.db_repo_name, commit_id=prev_commit.raw_id, f_path=f_path) if c.branch: c.url_prev += f'?branch={c.branch}' except (CommitDoesNotExistError, VCSError): c.url_prev = '#' c.prev_commit = EmptyCommit() # next link try: next_commit = c.commit.next(c.branch) c.next_commit = next_commit c.url_next = h.route_path('repo_files', repo_name=self.db_repo_name, commit_id=next_commit.raw_id, f_path=f_path) if c.branch: c.url_next += f'?branch={c.branch}' except (CommitDoesNotExistError, VCSError): c.url_next = '#' c.next_commit = EmptyCommit() # load file content if c.file.is_file(): c.lf_node = {} has_lf_enabled = self._is_lf_enabled(self.db_repo) if has_lf_enabled: c.lf_node = c.file.get_largefile_node() c.file_source_page = 'true' c.file_last_commit = c.file.last_commit c.file_size_too_big = c.file.size > c.visual.cut_off_limit_file if not (c.file_size_too_big or c.file.is_binary): if c.annotate: # annotation has precedence over renderer c.annotated_lines = filenode_as_annotated_lines_tokens( c.file ) else: c.renderer = ( c.renderer and h.renderer_from_filename(c.file.path) ) if not c.renderer: c.lines = filenode_as_lines_tokens(c.file) _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) c.on_branch_head = is_head branch = c.commit.branch if ( c.commit.branch and '/' not in c.commit.branch) else None c.branch_or_raw_id = branch or c.commit.raw_id c.branch_name = c.commit.branch or h.short_id(c.commit.raw_id) author = c.file_last_commit.author c.authors = [[ h.email(author), h.person(author, 'username_or_name_or_email'), 1 ]] else: # load tree content at path c.file_source_page = 'false' c.authors = [] # this loads a simple tree without metadata to speed things up # later via ajax we call repo_nodetree_full and fetch whole c.file_tree = self._get_tree_at_commit(c, c.commit.raw_id, f_path, at_rev=at_rev) c.readme_data, c.readme_file = \ self._get_readme_data(self.db_repo, c.visual.default_renderer, c.commit.raw_id, bytes_path) except RepositoryError as e: h.flash(h.escape(safe_str(e)), category='error') raise HTTPNotFound() if self.request.environ.get('HTTP_X_PJAX'): html = render('rhodecode:templates/files/files_pjax.mako', self._get_template_context(c), self.request) else: html = render('rhodecode:templates/files/files.mako', self._get_template_context(c), self.request) return Response(html) @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_files_annotated_previous(self): self.load_default_context() commit_id, bytes_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) prev_commit_id = commit.raw_id line_anchor = self.request.GET.get('line_anchor') is_file = False try: _file = commit.get_node(bytes_path) is_file = _file.is_file() except (NodeDoesNotExistError, CommitDoesNotExistError, VCSError): pass if is_file: history = commit.get_path_history(bytes_path) prev_commit_id = history[1].raw_id \ if len(history) > 1 else prev_commit_id prev_url = h.route_path( 'repo_files:annotated', repo_name=self.db_repo_name, commit_id=prev_commit_id, f_path=bytes_path, _anchor=f'L{line_anchor}') raise HTTPFound(prev_url) @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_nodetree_full(self): """ Returns rendered html of file tree that contains commit date, author, commit_id for the specified combination of repo, commit_id and file path """ c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) try: dir_node = commit.get_node(bytes_path) except RepositoryError as e: return Response(f'error: {h.escape(safe_str(e))}') if dir_node.is_file(): return Response('') c.file = dir_node c.commit = commit at_rev = self.request.GET.get('at') html = self._get_tree_at_commit( c, commit.raw_id, dir_node.path, full_load=True, at_rev=at_rev) return Response(html) def _get_attachement_headers(self, f_path): f_name = safe_str(f_path.split(Repository.NAME_SEP)[-1]) safe_path = f_name.replace('"', '\\"') encoded_path = urllib.parse.quote(f_name) headers = f"attachment; " \ f"filename=\"{safe_path}\"; " \ f"filename*=UTF-8\'\'{encoded_path}" return header_safe_str(headers) @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_file_raw(self): """ Action for show as raw, some mimetypes are "rendered", those include images, icons. """ c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) file_node = self._get_filenode_or_redirect(commit, bytes_path) raw_mimetype_mapping = { # map original mimetype to a mimetype used for "show as raw" # you can also provide a content-disposition to override the # default "attachment" disposition. # orig_type: (new_type, new_dispo) # show images inline: # Do not re-add SVG: it is unsafe and permits XSS attacks. One can # for example render an SVG with javascript inside or even render # HTML. 'image/x-icon': ('image/x-icon', 'inline'), 'image/png': ('image/png', 'inline'), 'image/gif': ('image/gif', 'inline'), 'image/jpeg': ('image/jpeg', 'inline'), 'application/pdf': ('application/pdf', 'inline'), } mimetype = file_node.mimetype try: mimetype, disposition = raw_mimetype_mapping[mimetype] except KeyError: # we don't know anything special about this, handle it safely if file_node.is_binary: # do same as download raw for binary files mimetype, disposition = 'application/octet-stream', 'attachment' else: # do not just use the original mimetype, but force text/plain, # otherwise it would serve text/html and that might be unsafe. # Note: underlying vcs library fakes text/plain mimetype if the # mimetype can not be determined and it thinks it is not # binary.This might lead to erroneous text display in some # cases, but helps in other cases, like with text files # without extension. mimetype, disposition = 'text/plain', 'inline' if disposition == 'attachment': disposition = self._get_attachement_headers(f_path) stream_content = file_node.stream_bytes() response = Response(app_iter=stream_content) response.content_disposition = disposition response.content_type = mimetype charset = self._get_default_encoding(c) if charset: response.charset = charset return response @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_file_download(self): c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) file_node = self._get_filenode_or_redirect(commit, bytes_path) if self.request.GET.get('lf'): # only if lf get flag is passed, we download this file # as LFS/Largefile lf_node = file_node.get_largefile_node() if lf_node: # overwrite our pointer with the REAL large-file file_node = lf_node disposition = self._get_attachement_headers(f_path) stream_content = file_node.stream_bytes() response = Response(app_iter=stream_content) response.content_disposition = disposition response.content_type = file_node.mimetype charset = self._get_default_encoding(c) if charset: response.charset = charset return response def _get_nodelist_at_commit(self, repo_name, repo_id, commit_id, f_path): cache_seconds = rhodecode.ConfigGet().get_int('rc_cache.cache_repo.expiration_time') cache_on = cache_seconds > 0 log.debug( 'Computing FILE SEARCH for repo_id %s commit_id `%s` and path `%s`' 'with caching: %s[TTL: %ss]' % ( repo_id, commit_id, f_path, cache_on, cache_seconds or 0)) cache_namespace_uid = f'repo.{repo_id}' region = rc_cache.get_or_create_region('cache_repo', cache_namespace_uid) @region.conditional_cache_on_arguments(namespace=cache_namespace_uid, condition=cache_on) def compute_file_search(_name_hash, _repo_id, _commit_id, _f_path): log.debug('Generating cached nodelist for repo_id:%s, %s, %s', _repo_id, commit_id, f_path) try: _d, _f = ScmModel().get_quick_filter_nodes(repo_name, _commit_id, _f_path) except (RepositoryError, CommitDoesNotExistError, Exception) as e: log.exception(safe_str(e)) h.flash(h.escape(safe_str(e)), category='error') raise HTTPFound(h.route_path('repo_files', repo_name=self.db_repo_name, commit_id='tip', f_path='/')) return _d + _f result = compute_file_search(self.db_repo.repo_name_hash, self.db_repo.repo_id, commit_id, f_path) return filter(lambda n: self.path_filter.path_access_allowed(n['name']), result) @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_nodelist(self): self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) metadata = self._get_nodelist_at_commit( self.db_repo_name, self.db_repo.repo_id, commit.raw_id, f_path) return {'nodes': [x for x in metadata]} def _create_references(self, branches_or_tags, symbolic_reference, f_path, ref_type): items = [] for name, commit_id in branches_or_tags.items(): sym_ref = symbolic_reference(commit_id, name, f_path, ref_type) items.append((sym_ref, name, ref_type)) return items def _symbolic_reference(self, commit_id, name, f_path, ref_type): return commit_id def _symbolic_reference_svn(self, commit_id, name, f_path, ref_type): return commit_id # NOTE(dan): old code we used in "diff" mode compare new_f_path = vcspath.join(name, f_path) return f'{new_f_path}@{commit_id}' def _get_node_history(self, commit_obj, f_path, commits=None): """ get commit history for given node :param commit_obj: commit to calculate history :param f_path: path for node to calculate history for :param commits: if passed don't calculate history and take commits defined in this list """ _ = self.request.translate # calculate history based on tip tip = self.rhodecode_vcs_repo.get_commit() if commits is None: pre_load = ["author", "branch"] try: commits = tip.get_path_history(safe_bytes(f_path), pre_load=pre_load) except (NodeDoesNotExistError, CommitError): # this node is not present at tip! commits = commit_obj.get_path_history(safe_bytes(f_path), pre_load=pre_load) history = [] commits_group = ([], _("Changesets")) for commit in commits: branch = ' (%s)' % commit.branch if commit.branch else '' n_desc = f'r{commit.idx}:{commit.short_id}{branch}' commits_group[0].append((commit.raw_id, n_desc, 'sha')) history.append(commits_group) symbolic_reference = self._symbolic_reference if self.rhodecode_vcs_repo.alias == 'svn': adjusted_f_path = RepoFilesView.adjust_file_path_for_svn( f_path, self.rhodecode_vcs_repo) if adjusted_f_path != f_path: log.debug( 'Recognized svn tag or branch in file "%s", using svn ' 'specific symbolic references', f_path) f_path = adjusted_f_path symbolic_reference = self._symbolic_reference_svn branches = self._create_references( self.rhodecode_vcs_repo.branches, symbolic_reference, f_path, 'branch') branches_group = (branches, _("Branches")) tags = self._create_references( self.rhodecode_vcs_repo.tags, symbolic_reference, f_path, 'tag') tags_group = (tags, _("Tags")) history.append(branches_group) history.append(tags_group) return history, commits @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_file_history(self): self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) file_node = self._get_filenode_or_redirect(commit, bytes_path) if file_node.is_file(): file_history, _hist = self._get_node_history(commit, f_path) res = [] for section_items, section in file_history: items = [] for obj_id, obj_text, obj_type in section_items: at_rev = '' if obj_type in ['branch', 'bookmark', 'tag']: at_rev = obj_text entry = { 'id': obj_id, 'text': obj_text, 'type': obj_type, 'at_rev': at_rev } items.append(entry) res.append({ 'text': section, 'children': items }) data = { 'more': False, 'results': res } return data log.warning('Cannot fetch history for directory') raise HTTPBadRequest() @LoginRequired() @HasRepoPermissionAnyDecorator( 'repository.read', 'repository.write', 'repository.admin') def repo_file_authors(self): c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() commit = self._get_commit_or_redirect(commit_id) file_node = self._get_filenode_or_redirect(commit, bytes_path) if not file_node.is_file(): raise HTTPBadRequest() c.file_last_commit = file_node.last_commit if self.request.GET.get('annotate') == '1': # use _hist from annotation if annotation mode is on commit_ids = {x[1] for x in file_node.annotate} _hist = ( self.rhodecode_vcs_repo.get_commit(commit_id) for commit_id in commit_ids) else: _f_history, _hist = self._get_node_history(commit, f_path) c.file_author = False unique = collections.OrderedDict() for commit in _hist: author = commit.author if author not in unique: unique[commit.author] = [ h.email(author), h.person(author, 'username_or_name_or_email'), 1 # counter ] else: # increase counter unique[commit.author][2] += 1 c.authors = [val for val in unique.values()] return self._get_template_context(c) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') def repo_files_check_head(self): self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) new_path = self.request.POST.get('path') operation = self.request.POST.get('operation') path_exist = '' if new_path and operation in ['create', 'upload']: new_f_path = os.path.join(f_path.lstrip('/'), new_path) try: commit_obj = self.rhodecode_vcs_repo.get_commit(commit_id) # NOTE(dan): construct whole path without leading / file_node = commit_obj.get_node(safe_bytes(new_f_path)) if file_node: path_exist = new_f_path except (EmptyRepositoryError, NodeDoesNotExistError): pass return { 'branch': _branch_name, 'sha': _sha_commit_id, 'is_head': is_head, 'path_exists': path_exist } @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') def repo_files_remove_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path) self.check_branch_permission(_branch_name) c.commit = self._get_commit_or_redirect(commit_id) c.file = self._get_filenode_or_redirect(c.commit, bytes_path) c.default_message = _( 'Deleted file {} via RhodeCode Enterprise').format(f_path) c.f_path = f_path return self._get_template_context(c) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') @CSRFRequired() def repo_files_delete_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path) self.check_branch_permission(_branch_name) c.commit = self._get_commit_or_redirect(commit_id) c.file = self._get_filenode_or_redirect(c.commit, bytes_path) c.default_message = _('Deleted file {} via RhodeCode Enterprise').format(f_path) c.f_path = f_path node_path = f_path author = self._rhodecode_db_user.full_contact message = self.request.POST.get('message') or c.default_message try: nodes = { safe_bytes(node_path): { 'content': b'' } } ScmModel().delete_nodes( user=self._rhodecode_db_user.user_id, repo=self.db_repo, message=message, nodes=nodes, parent_commit=c.commit, author=author, ) h.flash( _('Successfully deleted file `{}`').format( h.escape(f_path)), category='success') except Exception: log.exception('Error during commit operation') h.flash(_('Error occurred during commit'), category='error') raise HTTPFound( h.route_path('repo_commit', repo_name=self.db_repo_name, commit_id='tip')) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') def repo_files_edit_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path, commit_id=commit_id) self.check_branch_permission(_branch_name, commit_id=commit_id) c.commit = self._get_commit_or_redirect(commit_id) c.file = self._get_filenode_or_redirect(c.commit, bytes_path) if c.file.is_binary: files_url = h.route_path( 'repo_files', repo_name=self.db_repo_name, commit_id=c.commit.raw_id, f_path=f_path) raise HTTPFound(files_url) c.default_message = _('Edited file {} via RhodeCode Enterprise').format(f_path) c.f_path = f_path return self._get_template_context(c) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') @CSRFRequired() def repo_files_update_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() c.commit = self._get_commit_or_redirect(commit_id) c.file = self._get_filenode_or_redirect(c.commit, bytes_path) if c.file.is_binary: raise HTTPFound(h.route_path('repo_files', repo_name=self.db_repo_name, commit_id=c.commit.raw_id, f_path=f_path)) _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path, commit_id=commit_id) self.check_branch_permission(_branch_name, commit_id=commit_id) c.default_message = _('Edited file {} via RhodeCode Enterprise').format(f_path) c.f_path = f_path old_content = c.file.str_content sl = old_content.splitlines(1) first_line = sl[0] if sl else '' r_post = self.request.POST # line endings: 0 - Unix, 1 - Mac, 2 - DOS line_ending_mode = detect_mode(first_line, 0) content = convert_line_endings(r_post.get('content', ''), line_ending_mode) message = r_post.get('message') or c.default_message org_node_path = c.file.str_path filename = r_post['filename'] root_path = c.file.dir_path pure_path = self.create_pure_path(root_path, filename) node_path = pure_path.as_posix() default_redirect_url = h.route_path('repo_commit', repo_name=self.db_repo_name, commit_id=commit_id) if content == old_content and node_path == org_node_path: h.flash(_('No changes detected on {}').format(h.escape(org_node_path)), category='warning') raise HTTPFound(default_redirect_url) try: mapping = { c.file.bytes_path: { 'org_filename': org_node_path, 'filename': safe_bytes(node_path), 'content': safe_bytes(content), 'lexer': '', 'op': 'mod', 'mode': c.file.mode } } commit = ScmModel().update_nodes( user=self._rhodecode_db_user.user_id, repo=self.db_repo, message=message, nodes=mapping, parent_commit=c.commit, ) h.flash(_('Successfully committed changes to file `{}`').format( h.escape(f_path)), category='success') default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id) except Exception: log.exception('Error occurred during commit') h.flash(_('Error occurred during commit'), category='error') raise HTTPFound(default_redirect_url) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') def repo_files_add_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() # Check if we need to use this page to upload binary upload_binary = str2bool(self.request.params.get('upload_binary', False)) c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False) if c.commit is None: c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias) if self.rhodecode_vcs_repo.is_empty(): # for empty repository we cannot check for current branch, we rely on # c.commit.branch instead _branch_name, _sha_commit_id, is_head = c.commit.branch, '', True else: _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path, commit_id=commit_id) self.check_branch_permission(_branch_name, commit_id=commit_id) c.default_message = (_('Added file via RhodeCode Enterprise')) \ if not upload_binary else (_('Edited file {} via RhodeCode Enterprise').format(f_path)) c.f_path = f_path.lstrip('/') # ensure not relative path c.replace_binary = upload_binary return self._get_template_context(c) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') @CSRFRequired() def repo_files_create_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False) if c.commit is None: c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias) # calculate redirect URL if self.rhodecode_vcs_repo.is_empty(): default_redirect_url = h.route_path( 'repo_summary', repo_name=self.db_repo_name) else: default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id='tip') if self.rhodecode_vcs_repo.is_empty(): # for empty repository we cannot check for current branch, we rely on # c.commit.branch instead _branch_name, _sha_commit_id, is_head = c.commit.branch, '', True else: _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) self.forbid_non_head(is_head, f_path, commit_id=commit_id) self.check_branch_permission(_branch_name, commit_id=commit_id) c.default_message = (_('Added file via RhodeCode Enterprise')) c.f_path = f_path r_post = self.request.POST message = r_post.get('message') or c.default_message filename = r_post.get('filename') unix_mode = 0 if not filename: # If there's no commit, redirect to repo summary if type(c.commit) is EmptyCommit: redirect_url = h.route_path( 'repo_summary', repo_name=self.db_repo_name) else: redirect_url = default_redirect_url h.flash(_('No filename specified'), category='warning') raise HTTPFound(redirect_url) root_path = f_path pure_path = self.create_pure_path(root_path, filename) node_path = pure_path.as_posix().lstrip('/') author = self._rhodecode_db_user.full_contact content = convert_line_endings(r_post.get('content', ''), unix_mode) nodes = { safe_bytes(node_path): { 'content': safe_bytes(content) } } try: commit = ScmModel().create_nodes( user=self._rhodecode_db_user.user_id, repo=self.db_repo, message=message, nodes=nodes, parent_commit=c.commit, author=author, ) h.flash(_('Successfully committed new file `{}`').format(h.escape(node_path)), category='success') default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id) except NonRelativePathError: log.exception('Non Relative path found') h.flash(_('The location specified must be a relative path and must not ' 'contain .. in the path'), category='warning') raise HTTPFound(default_redirect_url) except (NodeError, NodeAlreadyExistsError) as e: h.flash(h.escape(safe_str(e)), category='error') except Exception: log.exception('Error occurred during commit') h.flash(_('Error occurred during commit'), category='error') raise HTTPFound(default_redirect_url) @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') @CSRFRequired() def repo_files_upload_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False) if c.commit is None: c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias) # calculate redirect URL if self.rhodecode_vcs_repo.is_empty(): default_redirect_url = h.route_path( 'repo_summary', repo_name=self.db_repo_name) else: default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id='tip') if self.rhodecode_vcs_repo.is_empty(): # for empty repository we cannot check for current branch, we rely on # c.commit.branch instead _branch_name, _sha_commit_id, is_head = c.commit.branch, '', True else: _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) error = self.forbid_non_head(is_head, f_path, json_mode=True) if error: return { 'error': error, 'redirect_url': default_redirect_url } error = self.check_branch_permission(_branch_name, json_mode=True) if error: return { 'error': error, 'redirect_url': default_redirect_url } c.default_message = (_('Added file via RhodeCode Enterprise')) c.f_path = f_path r_post = self.request.POST message = c.default_message user_message = r_post.getall('message') if isinstance(user_message, list) and user_message: # we take the first from duplicated results if it's not empty message = user_message[0] if user_message[0] else message nodes = {} for file_obj in r_post.getall('files_upload') or []: content = file_obj.file filename = file_obj.filename root_path = f_path pure_path = self.create_pure_path(root_path, filename) node_path = pure_path.as_posix().lstrip('/') nodes[safe_bytes(node_path)] = { 'content': content } if not nodes: error = 'missing files' return { 'error': error, 'redirect_url': default_redirect_url } author = self._rhodecode_db_user.full_contact try: commit = ScmModel().create_nodes( user=self._rhodecode_db_user.user_id, repo=self.db_repo, message=message, nodes=nodes, parent_commit=c.commit, author=author, ) if len(nodes) == 1: flash_message = _('Successfully committed {} new files').format(len(nodes)) else: flash_message = _('Successfully committed 1 new file') h.flash(flash_message, category='success') default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id) except NonRelativePathError: log.exception('Non Relative path found') error = _('The location specified must be a relative path and must not ' 'contain .. in the path') h.flash(error, category='warning') return { 'error': error, 'redirect_url': default_redirect_url } except (NodeError, NodeAlreadyExistsError) as e: error = h.escape(e) h.flash(error, category='error') return { 'error': error, 'redirect_url': default_redirect_url } except Exception: log.exception('Error occurred during commit') error = _('Error occurred during commit') h.flash(error, category='error') return { 'error': error, 'redirect_url': default_redirect_url } return { 'error': None, 'redirect_url': default_redirect_url } @LoginRequired() @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin') @CSRFRequired() def repo_files_replace_file(self): _ = self.request.translate c = self.load_default_context() commit_id, f_path, bytes_path = self._get_commit_and_path() self._ensure_not_locked() c.commit = self._get_commit_or_redirect(commit_id, redirect_after=False) if c.commit is None: c.commit = EmptyCommit(alias=self.rhodecode_vcs_repo.alias) if self.rhodecode_vcs_repo.is_empty(): default_redirect_url = h.route_path( 'repo_summary', repo_name=self.db_repo_name) else: default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id='tip') if self.rhodecode_vcs_repo.is_empty(): # for empty repository we cannot check for current branch, we rely on # c.commit.branch instead _branch_name, _sha_commit_id, is_head = c.commit.branch, '', True else: _branch_name, _sha_commit_id, is_head = \ self._is_valid_head(commit_id, self.rhodecode_vcs_repo, landing_ref=self.db_repo.landing_ref_name) error = self.forbid_non_head(is_head, f_path, json_mode=True) if error: return { 'error': error, 'redirect_url': default_redirect_url } error = self.check_branch_permission(_branch_name, json_mode=True) if error: return { 'error': error, 'redirect_url': default_redirect_url } c.default_message = (_('Edited file {} via RhodeCode Enterprise').format(f_path)) c.f_path = f_path r_post = self.request.POST message = c.default_message user_message = r_post.getall('message') if isinstance(user_message, list) and user_message: # we take the first from duplicated results if it's not empty message = user_message[0] if user_message[0] else message data_for_replacement = r_post.getall('files_upload') or [] if (objects_count := len(data_for_replacement)) > 1: return { 'error': 'too many files for replacement', 'redirect_url': default_redirect_url } elif not objects_count: return { 'error': 'missing files', 'redirect_url': default_redirect_url } content = data_for_replacement[0].file retrieved_filename = data_for_replacement[0].filename if retrieved_filename.split('.')[-1] != f_path.split('.')[-1]: return { 'error': 'file extension of uploaded file doesn\'t match an original file\'s extension', 'redirect_url': default_redirect_url } author = self._rhodecode_db_user.full_contact try: commit = ScmModel().update_binary_node( user=self._rhodecode_db_user.user_id, repo=self.db_repo, message=message, node={ 'content': content, 'file_path': f_path.encode(), }, parent_commit=c.commit, author=author, ) h.flash(_('Successfully committed 1 new file'), category='success') default_redirect_url = h.route_path( 'repo_commit', repo_name=self.db_repo_name, commit_id=commit.raw_id) except (NodeError, NodeAlreadyExistsError) as e: error = h.escape(e) h.flash(error, category='error') return { 'error': error, 'redirect_url': default_redirect_url } except Exception: log.exception('Error occurred during commit') error = _('Error occurred during commit') h.flash(error, category='error') return { 'error': error, 'redirect_url': default_redirect_url } return { 'error': None, 'redirect_url': default_redirect_url }