##// END OF EJS Templates
updated latest release notes.
updated latest release notes.

File last commit:

r5647:8333bc7b default
r5650:cb066911 default
Show More
repo_files.py
1704 lines | 64.6 KiB | text/x-python | PythonLexer
# 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 <http://www.gnu.org/licenses/>.
#
# 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
}