security: fixed self-xss inside file views.

rhodecode/controllers/files.py

@@ -1,1110 +1,1110 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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/
 
 """
 Files controller for RhodeCode Enterprise
 """
 
 import itertools
 import logging
 import os
 import shutil
 import tempfile
 
 from pylons import request, response, tmpl_context as c, url
 from pylons.i18n.translation import _
 from pylons.controllers.util import redirect
 from webob.exc import HTTPNotFound, HTTPBadRequest
 
 from rhodecode.controllers.utils import parse_path_ref
 from rhodecode.lib import diffs, helpers as h, caches
 from rhodecode.lib import audit_logger
 from rhodecode.lib.codeblocks import (
     filenode_as_lines_tokens, filenode_as_annotated_lines_tokens)
 from rhodecode.lib.utils import jsonify
 from rhodecode.lib.utils2 import (
     convert_line_endings, detect_mode, safe_str, str2bool)
 from rhodecode.lib.auth import (
     LoginRequired, HasRepoPermissionAnyDecorator, CSRFRequired, XHRRequired)
 from rhodecode.lib.base import BaseRepoController, render
 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.exceptions import (
     RepositoryError, CommitDoesNotExistError, EmptyRepositoryError,
     ImproperArchiveTypeError, VCSError, NodeAlreadyExistsError,
     NodeDoesNotExistError, CommitError, NodeError)
 from rhodecode.lib.vcs.nodes import FileNode
 
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.db import Repository
 
 from rhodecode.controllers.changeset import (
     _ignorews_url, _context_url, get_line_ctx, get_ignore_ws)
 from rhodecode.lib.exceptions import NonRelativePathError
 
 log = logging.getLogger(__name__)
 
 
 class FilesController(BaseRepoController):
 
     def __before__(self):
         super(FilesController, self).__before__()
         c.cut_off_limit = self.cut_off_limit_file
 
     def _get_default_encoding(self):
         enc_list = getattr(c, 'default_encodings', [])
         return enc_list[0] if enc_list else 'UTF-8'
 
     def __get_commit_or_redirect(self, commit_id, repo_name,
                                  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 repo_name: repo name to redirect after
         :param redirect_after: toggle redirection
         """
         try:
             return c.rhodecode_repo.get_commit(commit_id)
         except EmptyRepositoryError:
             if not redirect_after:
                 return None
             url_ = url('files_add_home',
                        repo_name=c.repo_name,
                        revision=0, f_path='', anchor='edit')
             if h.HasRepoPermissionAny(
                     'repository.write', 'repository.admin')(c.repo_name):
                 add_new = h.link_to(
                     _('Click here to add a new file.'),
                     url_, class_="alert-link")
             else:
                 add_new = ""
             h.flash(h.literal(
                 _('There are no files yet. %s') % add_new), category='warning')
             redirect(h.route_path('repo_summary', repo_name=repo_name))
         except (CommitDoesNotExistError, LookupError):
             msg = _('No such commit exists for this repository')
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
 
     def __get_filenode_or_redirect(self, repo_name, commit, path):
         """
         Returns file_node, if error occurs or given path is directory,
         it'll redirect to top level path
 
         :param repo_name: repo_name
         :param commit: given commit
         :param path: path to lookup
         """
         try:
             file_node = commit.get_node(path)
             if file_node.is_dir():
                 raise RepositoryError('The given path is a directory')
         except CommitDoesNotExistError:
-            msg = _('No such commit exists for this repository')
-            log.exception(msg)
-            h.flash(msg, category='error')
+            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:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
 
         return file_node
 
     def __get_tree_cache_manager(self, repo_name, namespace_type):
         _namespace = caches.get_repo_namespace_key(namespace_type, repo_name)
         return caches.get_cache_manager('repo_cache_long', _namespace)
 
     def _get_tree_at_commit(self, repo_name, commit_id, f_path,
                             full_load=False, force=False):
         def _cached_tree():
             log.debug('Generating cached file tree for %s, %s, %s',
                       repo_name, commit_id, f_path)
             c.full_load = full_load
             return render('files/files_browser_tree.mako')
 
         cache_manager = self.__get_tree_cache_manager(
             repo_name, caches.FILE_TREE)
 
         cache_key = caches.compute_key_from_params(
             repo_name, commit_id, f_path)
 
         if force:
             # we want to force recompute of caches
             cache_manager.remove_value(cache_key)
 
         return cache_manager.get(cache_key, createfunc=_cached_tree)
 
     def _get_nodelist_at_commit(self, repo_name, commit_id, f_path):
         def _cached_nodes():
             log.debug('Generating cached nodelist for %s, %s, %s',
                       repo_name, commit_id, f_path)
             _d, _f = ScmModel().get_nodes(
                 repo_name, commit_id, f_path, flat=False)
             return _d + _f
 
         cache_manager = self.__get_tree_cache_manager(
             repo_name, caches.FILE_SEARCH_TREE_META)
 
         cache_key = caches.compute_key_from_params(
             repo_name, commit_id, f_path)
         return cache_manager.get(cache_key, createfunc=_cached_nodes)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def index(
             self, repo_name, revision, f_path, annotate=False, rendered=False):
         commit_id = revision
 
         # redirect to given commit_id from form if given
         get_commit_id = request.GET.get('at_rev', None)
         if get_commit_id:
             self.__get_commit_or_redirect(get_commit_id, repo_name)
 
         c.commit = self.__get_commit_or_redirect(commit_id, repo_name)
         c.branch = request.GET.get('branch', None)
         c.f_path = f_path
         c.annotate = annotate
         # default is false, but .rst/.md files later are autorendered, we can
         # overwrite autorendering by setting this GET flag
         c.renderer = rendered or not request.GET.get('no-render', False)
 
         # prev link
         try:
             prev_commit = c.commit.prev(c.branch)
             c.prev_commit = prev_commit
             c.url_prev = url('files_home', repo_name=c.repo_name,
                              revision=prev_commit.raw_id, f_path=f_path)
             if c.branch:
                 c.url_prev += '?branch=%s' % 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 = url('files_home', repo_name=c.repo_name,
                              revision=next_commit.raw_id, f_path=f_path)
             if c.branch:
                 c.url_next += '?branch=%s' % c.branch
         except (CommitDoesNotExistError, VCSError):
             c.url_next = '#'
             c.next_commit = EmptyCommit()
 
         # files or dirs
         try:
             c.file = c.commit.get_node(f_path)
             c.file_author = True
             c.file_tree = ''
             if c.file.is_file():
                 c.lf_node = c.file.get_largefile_node()
 
                 c.file_source_page = 'true'
                 c.file_last_commit = c.file.last_commit
                 if c.file.size < self.cut_off_limit_file:
                     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)
 
                 c.on_branch_head = self._is_valid_head(
                     commit_id, c.rhodecode_repo)
 
                 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'))]
             else:
                 c.file_source_page = 'false'
                 c.authors = []
                 c.file_tree = self._get_tree_at_commit(
                     repo_name, c.commit.raw_id, f_path)
 
         except RepositoryError as e:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
 
         if request.environ.get('HTTP_X_PJAX'):
             return render('files/files_pjax.mako')
 
         return render('files/files.mako')
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def annotate_previous(self, repo_name, revision, f_path):
 
         commit_id = revision
         commit = self.__get_commit_or_redirect(commit_id, repo_name)
         prev_commit_id = commit.raw_id
 
         f_path = f_path
         is_file = False
         try:
             _file = commit.get_node(f_path)
             is_file = _file.is_file()
         except (NodeDoesNotExistError, CommitDoesNotExistError, VCSError):
             pass
 
         if is_file:
             history = commit.get_file_history(f_path)
             prev_commit_id = history[1].raw_id \
                 if len(history) > 1 else prev_commit_id
 
         return redirect(h.url(
             'files_annotate_home', repo_name=repo_name,
             revision=prev_commit_id, f_path=f_path))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     @jsonify
     def history(self, repo_name, revision, f_path):
         commit = self.__get_commit_or_redirect(revision, repo_name)
         f_path = f_path
         _file = commit.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(commit, f_path)
 
             res = []
             for obj in file_history:
                 res.append({
                     'text': obj[1],
                     'children': [{'id': o[0], 'text': o[1]} for o in obj[0]]
                 })
 
             data = {
                 'more': False,
                 'results': res
             }
             return data
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def authors(self, repo_name, revision, f_path):
         commit = self.__get_commit_or_redirect(revision, repo_name)
         file_node = commit.get_node(f_path)
         if file_node.is_file():
             c.file_last_commit = file_node.last_commit
             if request.GET.get('annotate') == '1':
                 # use _hist from annotation if annotation mode is on
                 commit_ids = set(x[1] for x in file_node.annotate)
                 _hist = (
                     c.rhodecode_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
             c.authors = []
             for author in set(commit.author for commit in _hist):
                 c.authors.append((
                     h.email(author),
                     h.person(author, 'username_or_name_or_email')))
             return render('files/file_authors_box.mako')
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def rawfile(self, repo_name, revision, f_path):
         """
         Action for download as raw
         """
         commit = self.__get_commit_or_redirect(revision, repo_name)
         file_node = self.__get_filenode_or_redirect(repo_name, commit, f_path)
 
         if 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
 
         response.content_disposition = 'attachment; filename=%s' % \
             safe_str(f_path.split(Repository.NAME_SEP)[-1])
 
         response.content_type = file_node.mimetype
         charset = self._get_default_encoding()
         if charset:
             response.charset = charset
 
         return file_node.content
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def raw(self, repo_name, revision, f_path):
         """
         Action for show as raw, some mimetypes are "rendered",
         those include images, icons.
         """
         commit = self.__get_commit_or_redirect(revision, repo_name)
         file_node = self.__get_filenode_or_redirect(repo_name, commit, f_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, dispo = 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, dispo = '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, dispo = 'text/plain', 'inline'
 
         if dispo == 'attachment':
             dispo = 'attachment; filename=%s' % safe_str(
                 f_path.split(os.sep)[-1])
 
         response.content_disposition = dispo
         response.content_type = mimetype
         charset = self._get_default_encoding()
         if charset:
             response.charset = charset
         return file_node.content
 
     @CSRFRequired()
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def delete(self, repo_name, revision, f_path):
         commit_id = revision
 
         repo = c.rhodecode_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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         if not self._is_valid_head(commit_id, repo.scm_instance()):
             h.flash(_('You can only delete files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
 
         c.commit = self.__get_commit_or_redirect(commit_id, repo_name)
         c.file = self.__get_filenode_or_redirect(repo_name, c.commit, f_path)
 
         c.default_message = _(
-            'Deleted file %s via RhodeCode Enterprise') % (f_path)
+            'Deleted file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
         node_path = f_path
         author = c.rhodecode_user.full_contact
         message = request.POST.get('message') or c.default_message
         try:
             nodes = {
                 node_path: {
                     'content': ''
                 }
             }
             self.scm_model.delete_nodes(
                 user=c.rhodecode_user.user_id, repo=c.rhodecode_db_repo,
                 message=message,
                 nodes=nodes,
                 parent_commit=c.commit,
                 author=author,
             )
 
-            h.flash(_('Successfully deleted file %s') % f_path,
-                    category='success')
+            h.flash(
+                _('Successfully deleted file `{}`').format(
+                    h.escape(f_path)), category='success')
         except Exception:
             msg = _('Error occurred during commit')
             log.exception(msg)
             h.flash(msg, category='error')
         return redirect(url('changeset_home',
                             repo_name=c.repo_name, revision='tip'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def delete_home(self, repo_name, revision, f_path):
         commit_id = revision
 
         repo = c.rhodecode_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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         if not self._is_valid_head(commit_id, repo.scm_instance()):
             h.flash(_('You can only delete files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
 
         c.commit = self.__get_commit_or_redirect(commit_id, repo_name)
         c.file = self.__get_filenode_or_redirect(repo_name, c.commit, f_path)
 
         c.default_message = _(
-            'Deleted file %s via RhodeCode Enterprise') % (f_path)
+            'Deleted file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
 
         return render('files/files_delete.mako')
 
     @CSRFRequired()
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def edit(self, repo_name, revision, f_path):
         commit_id = revision
 
         repo = c.rhodecode_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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         if not self._is_valid_head(commit_id, repo.scm_instance()):
             h.flash(_('You can only edit files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
 
         c.commit = self.__get_commit_or_redirect(commit_id, repo_name)
         c.file = self.__get_filenode_or_redirect(repo_name, c.commit, f_path)
 
         if c.file.is_binary:
             return redirect(url('files_home', repo_name=c.repo_name,
                             revision=c.commit.raw_id, f_path=f_path))
         c.default_message = _(
-            'Edited file %s via RhodeCode Enterprise') % (f_path)
+            'Edited file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
         old_content = c.file.content
         sl = old_content.splitlines(1)
         first_line = sl[0] if sl else ''
 
         # modes:  0 - Unix, 1 - Mac, 2 - DOS
         mode = detect_mode(first_line, 0)
         content = convert_line_endings(request.POST.get('content', ''), mode)
 
         message = request.POST.get('message') or c.default_message
         org_f_path = c.file.unicode_path
         filename = request.POST['filename']
         org_filename = c.file.name
 
         if content == old_content and filename == org_filename:
             h.flash(_('No changes'), category='warning')
             return redirect(url('changeset_home', repo_name=c.repo_name,
                                 revision='tip'))
         try:
             mapping = {
                 org_f_path: {
                     'org_filename': org_f_path,
                     'filename': os.path.join(c.file.dir_path, filename),
                     'content': content,
                     'lexer': '',
                     'op': 'mod',
                 }
             }
 
             ScmModel().update_nodes(
                 user=c.rhodecode_user.user_id,
                 repo=c.rhodecode_db_repo,
                 message=message,
                 nodes=mapping,
                 parent_commit=c.commit,
             )
 
-            h.flash(_('Successfully committed to %s') % f_path,
-                    category='success')
+            h.flash(
+                _('Successfully committed changes to file `{}`').format(
+                    h.escape(f_path)), category='success')
         except Exception:
-            msg = _('Error occurred during commit')
-            log.exception(msg)
-            h.flash(msg, category='error')
+            log.exception('Error occurred during commit')
+            h.flash(_('Error occurred during commit'), category='error')
         return redirect(url('changeset_home',
                             repo_name=c.repo_name, revision='tip'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def edit_home(self, repo_name, revision, f_path):
         commit_id = revision
 
         repo = c.rhodecode_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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         if not self._is_valid_head(commit_id, repo.scm_instance()):
             h.flash(_('You can only edit files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
 
         c.commit = self.__get_commit_or_redirect(commit_id, repo_name)
         c.file = self.__get_filenode_or_redirect(repo_name, c.commit, f_path)
 
         if c.file.is_binary:
             return redirect(url('files_home', repo_name=c.repo_name,
                             revision=c.commit.raw_id, f_path=f_path))
         c.default_message = _(
-            'Edited file %s via RhodeCode Enterprise') % (f_path)
+            'Edited file {} via RhodeCode Enterprise').format(f_path)
         c.f_path = f_path
 
         return render('files/files_edit.mako')
 
     def _is_valid_head(self, commit_id, repo):
         # check if commit is a branch identifier- basically we cannot
         # create multiple heads via file editing
         valid_heads = repo.branches.keys() + repo.branches.values()
 
         if h.is_svn(repo) and not repo.is_empty():
             # Note: Subversion only has one head, we add it here in case there
             # is no branch matched.
             valid_heads.append(repo.get_commit(commit_idx=-1).raw_id)
 
         # check if commit is a branch name or branch hash
         return commit_id in valid_heads
 
     @CSRFRequired()
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def add(self, repo_name, revision, f_path):
         repo = Repository.get_by_repo_name(repo_name)
         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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         r_post = request.POST
 
         c.commit = self.__get_commit_or_redirect(
             revision, repo_name, redirect_after=False)
         if c.commit is None:
             c.commit = EmptyCommit(alias=c.rhodecode_repo.alias)
         c.default_message = (_('Added file via RhodeCode Enterprise'))
         c.f_path = f_path
         unix_mode = 0
         content = convert_line_endings(r_post.get('content', ''), unix_mode)
 
         message = r_post.get('message') or c.default_message
         filename = r_post.get('filename')
         location = r_post.get('location', '')  # dir location
         file_obj = r_post.get('upload_file', None)
 
         if file_obj is not None and hasattr(file_obj, 'filename'):
             filename = r_post.get('filename_upload')
             content = file_obj.file
 
             if hasattr(content, 'file'):
                 # non posix systems store real file under file attr
                 content = content.file
 
         # If there's no commit, redirect to repo summary
         if type(c.commit) is EmptyCommit:
             redirect_url = h.route_path('repo_summary', repo_name=c.repo_name)
         else:
             redirect_url = url("changeset_home", repo_name=c.repo_name,
                                revision='tip')
 
         if not filename:
             h.flash(_('No filename'), category='warning')
             return redirect(redirect_url)
 
         # extract the location from filename,
         # allows using foo/bar.txt syntax to create subdirectories
         subdir_loc = filename.rsplit('/', 1)
         if len(subdir_loc) == 2:
             location = os.path.join(location, subdir_loc[0])
 
         # strip all crap out of file, just leave the basename
         filename = os.path.basename(filename)
         node_path = os.path.join(location, filename)
         author = c.rhodecode_user.full_contact
 
         try:
             nodes = {
                 node_path: {
                     'content': content
                 }
             }
             self.scm_model.create_nodes(
                 user=c.rhodecode_user.user_id,
                 repo=c.rhodecode_db_repo,
                 message=message,
                 nodes=nodes,
                 parent_commit=c.commit,
                 author=author,
             )
 
-            h.flash(_('Successfully committed to %s') % node_path,
-                    category='success')
+            h.flash(
+                _('Successfully committed new file `{}`').format(
+                    h.escape(node_path)), category='success')
         except NonRelativePathError as e:
             h.flash(_(
                 'The location specified must be a relative path and must not '
                 'contain .. in the path'), category='warning')
             return redirect(url('changeset_home', repo_name=c.repo_name,
                                 revision='tip'))
         except (NodeError, NodeAlreadyExistsError) as e:
-            h.flash(_(e), category='error')
+            h.flash(_(h.escape(e)), category='error')
         except Exception:
-            msg = _('Error occurred during commit')
-            log.exception(msg)
-            h.flash(msg, category='error')
+            log.exception('Error occurred during commit')
+            h.flash(_('Error occurred during commit'), category='error')
         return redirect(url('changeset_home',
                             repo_name=c.repo_name, revision='tip'))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def add_home(self, repo_name, revision, f_path):
 
         repo = Repository.get_by_repo_name(repo_name)
         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')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
 
         c.commit = self.__get_commit_or_redirect(
             revision, repo_name, redirect_after=False)
         if c.commit is None:
             c.commit = EmptyCommit(alias=c.rhodecode_repo.alias)
         c.default_message = (_('Added file via RhodeCode Enterprise'))
         c.f_path = f_path
 
         return render('files/files_add.mako')
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def archivefile(self, repo_name, fname):
         fileformat = None
         commit_id = None
         ext = None
         subrepos = request.GET.get('subrepos') == 'true'
 
         for a_type, ext_data in settings.ARCHIVE_SPECS.items():
             archive_spec = fname.split(ext_data[1])
             if len(archive_spec) == 2 and archive_spec[1] == '':
                 fileformat = a_type or ext_data[1]
                 commit_id = archive_spec[0]
                 ext = ext_data[1]
 
         dbrepo = RepoModel().get_by_repo_name(repo_name)
         if not dbrepo.enable_downloads:
             return _('Downloads disabled')
 
         try:
             commit = c.rhodecode_repo.get_commit(commit_id)
             content_type = settings.ARCHIVE_SPECS[fileformat][0]
         except CommitDoesNotExistError:
             return _('Unknown revision %s') % commit_id
         except EmptyRepositoryError:
             return _('Empty repository')
         except KeyError:
             return _('Unknown archive type')
 
         # archive cache
         from rhodecode import CONFIG
 
         archive_name = '%s-%s%s%s' % (
             safe_str(repo_name.replace('/', '_')),
             '-sub' if subrepos else '',
             safe_str(commit.short_id), ext)
 
         use_cached_archive = False
         archive_cache_enabled = CONFIG.get(
             'archive_cache_dir') and not request.GET.get('no_cache')
 
         if archive_cache_enabled:
             # check if we it's ok to write
             if not os.path.isdir(CONFIG['archive_cache_dir']):
                 os.makedirs(CONFIG['archive_cache_dir'])
             cached_archive_path = os.path.join(
                 CONFIG['archive_cache_dir'], archive_name)
             if os.path.isfile(cached_archive_path):
                 log.debug('Found cached archive in %s', cached_archive_path)
                 fd, archive = None, cached_archive_path
                 use_cached_archive = True
             else:
                 log.debug('Archive %s is not yet cached', archive_name)
 
         if not use_cached_archive:
             # generate new archive
             fd, archive = tempfile.mkstemp()
-            log.debug('Creating new temp archive in %s' % (archive,))
+            log.debug('Creating new temp archive in %s', archive)
             try:
                 commit.archive_repo(archive, kind=fileformat, subrepos=subrepos)
             except ImproperArchiveTypeError:
                 return _('Unknown archive type')
             if archive_cache_enabled:
                 # if we generated the archive and we have cache enabled
                 # let's use this for future
-                log.debug('Storing new archive in %s' % (cached_archive_path,))
+                log.debug('Storing new archive in %s', cached_archive_path)
                 shutil.move(archive, cached_archive_path)
                 archive = cached_archive_path
 
         # store download action
         audit_logger.store_web(
             action='repo.archive.download',
             action_data={'user_agent': request.user_agent,
                          'archive_name': archive_name,
                          'archive_spec': fname,
                          'archive_cached': use_cached_archive},
             user=c.rhodecode_user,
             repo=dbrepo,
             commit=True
         )
 
         response.content_disposition = str(
             'attachment; filename=%s' % archive_name)
         response.content_type = str(content_type)
 
         def get_chunked_archive(archive):
             with open(archive, 'rb') as stream:
                 while True:
                     data = stream.read(16 * 1024)
                     if not data:
                         if fd:  # fd means we used temporary file
                             os.close(fd)
                         if not archive_cache_enabled:
                             log.debug('Destroying temp archive %s', archive)
                             os.remove(archive)
                         break
                     yield data
 
         return get_chunked_archive(archive)
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def diff(self, repo_name, f_path):
 
         c.action = request.GET.get('diff')
         diff1 = request.GET.get('diff1', '')
         diff2 = request.GET.get('diff2', '')
 
         path1, diff1 = parse_path_ref(diff1, default_path=f_path)
 
         ignore_whitespace = str2bool(request.GET.get('ignorews'))
         line_context = 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()
 
         if c.action not in ['download', 'raw']:
             # redirect to new view if we render diff
             return redirect(
                 url('compare_url', repo_name=repo_name,
                     source_ref_type='rev',
                     source_ref=diff1,
                     target_repo=c.repo_name,
                     target_ref_type='rev',
                     target_ref=diff2,
                     f_path=f_path))
 
         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")
             return redirect(url(
                 'files_home', repo_name=c.repo_name, 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)
             diff = diffs.DiffProcessor(_diff, format='gitdiff')
 
             diff_name = '%s_vs_%s.diff' % (diff1, diff2)
             response.content_type = 'text/plain'
             response.content_disposition = (
                 'attachment; filename=%s' % (diff_name,)
             )
             charset = self._get_default_encoding()
             if charset:
                 response.charset = charset
             return diff.as_raw()
 
         elif c.action == 'raw':
             _diff = diffs.get_gitdiff(node1, node2,
                                       ignore_whitespace=ignore_whitespace,
                                       context=line_context)
             diff = diffs.DiffProcessor(_diff, format='gitdiff')
             response.content_type = 'text/plain'
             charset = self._get_default_encoding()
             if charset:
                 response.charset = charset
             return diff.as_raw()
 
         else:
             return redirect(
                 url('compare_url', repo_name=repo_name,
                     source_ref_type='rev',
                     source_ref=diff1,
                     target_repo=c.repo_name,
                     target_ref_type='rev',
                     target_ref=diff2,
                     f_path=f_path))
 
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def diff_2way(self, repo_name, f_path):
         """
         Kept only to make OLD links work
         """
         diff1 = request.GET.get('diff1', '')
         diff2 = 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()
 
         return redirect(
             url('compare_url', repo_name=repo_name,
                 source_ref_type='rev',
                 source_ref=diff1,
                 target_repo=c.repo_name,
                 target_ref_type='rev',
                 target_ref=diff2,
                 f_path=f_path,
                 diffmode='sideside'))
 
     def _get_file_node(self, commit_id, f_path):
         if commit_id not in ['', None, 'None', '0' * 12, '0' * 40]:
             commit = c.rhodecode_repo.get_commit(commit_id=commit_id)
             try:
                 node = commit.get_node(f_path)
                 if node.is_dir():
                     raise NodeError('%s path is a %s not a file'
                                     % (node, type(node)))
             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(f_path, '', commit=commit)
         else:
             commit = EmptyCommit(
                 repo=c.rhodecode_repo,
                 alias=c.rhodecode_repo.alias)
             node = FileNode(f_path, '', commit=commit)
         return node
 
     def _get_node_history(self, commit, f_path, commits=None):
         """
         get commit history for given node
 
         :param commit: 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
         """
         # calculate history based on tip
         tip = c.rhodecode_repo.get_commit()
         if commits is None:
             pre_load = ["author", "branch"]
             try:
                 commits = tip.get_file_history(f_path, pre_load=pre_load)
             except (NodeDoesNotExistError, CommitError):
                 # this node is not present at tip!
                 commits = commit.get_file_history(f_path, pre_load=pre_load)
 
         history = []
         commits_group = ([], _("Changesets"))
         for commit in commits:
             branch = ' (%s)' % commit.branch if commit.branch else ''
             n_desc = 'r%s:%s%s' % (commit.idx, commit.short_id, branch)
             commits_group[0].append((commit.raw_id, n_desc,))
         history.append(commits_group)
 
         symbolic_reference = self._symbolic_reference
 
         if c.rhodecode_repo.alias == 'svn':
             adjusted_f_path = self._adjust_file_path_for_svn(
                 f_path, c.rhodecode_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(
             c.rhodecode_repo.branches, symbolic_reference, f_path)
         branches_group = (branches, _("Branches"))
 
         tags = self._create_references(
             c.rhodecode_repo.tags, symbolic_reference, f_path)
         tags_group = (tags, _("Tags"))
 
         history.append(branches_group)
         history.append(tags_group)
 
         return history, commits
 
     def _adjust_file_path_for_svn(self, 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.iterkeys(),
             repo.tags.iterkeys())
         tags_and_branches = sorted(tags_and_branches, key=len, reverse=True)
 
         for name in tags_and_branches:
             if f_path.startswith(name + '/'):
                 f_path = vcspath.relpath(f_path, name)
                 break
         return f_path
 
     def _create_references(
             self, branches_or_tags, symbolic_reference, f_path):
         items = []
         for name, commit_id in branches_or_tags.items():
             sym_ref = symbolic_reference(commit_id, name, f_path)
             items.append((sym_ref, name))
         return items
 
     def _symbolic_reference(self, commit_id, name, f_path):
         return commit_id
 
     def _symbolic_reference_svn(self, commit_id, name, f_path):
         new_f_path = vcspath.join(name, f_path)
         return u'%s@%s' % (new_f_path, commit_id)
 
     @LoginRequired()
     @XHRRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     @jsonify
     def nodelist(self, repo_name, revision, f_path):
         commit = self.__get_commit_or_redirect(revision, repo_name)
 
         metadata = self._get_nodelist_at_commit(
             repo_name, commit.raw_id, f_path)
         return {'nodes': metadata}
 
     @LoginRequired()
     @XHRRequired()
     @HasRepoPermissionAnyDecorator(
         'repository.read', 'repository.write', 'repository.admin')
     def nodetree_full(self, repo_name, commit_id, f_path):
         """
         Returns rendered html of file tree that contains commit date,
         author, revision for the specified combination of
         repo, commit_id and file path
 
         :param repo_name: name of the repository
         :param commit_id: commit_id of file tree
         :param f_path: file path of the requested directory
         """
 
         commit = self.__get_commit_or_redirect(commit_id, repo_name)
         try:
             dir_node = commit.get_node(f_path)
         except RepositoryError as e:
             return 'error {}'.format(safe_str(e))
 
         if dir_node.is_file():
             return ''
 
         c.file = dir_node
         c.commit = commit
 
         # using force=True here, make a little trick. We flush the cache and
         # compute it using the same key as without full_load, so the fully
         # loaded cached tree is now returned instead of partial
         return self._get_tree_at_commit(
             repo_name, commit.raw_id, dir_node.path, full_load=True,
             force=True)

rhodecode/tests/functional/test_files.py

@@ -1,988 +1,987 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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 os
 
 import mock
 import pytest
 
 from rhodecode.controllers.files import FilesController
 from rhodecode.lib import helpers as h
 from rhodecode.lib.compat import OrderedDict
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.vcs import nodes
 
 from rhodecode.lib.vcs.conf import settings
 from rhodecode.tests import (
     url, assert_session_flash, assert_not_in_session_flash)
 from rhodecode.tests.fixture import Fixture
 
 fixture = Fixture()
 
 NODE_HISTORY = {
     'hg': json.loads(fixture.load_resource('hg_node_history_response.json')),
     'git': json.loads(fixture.load_resource('git_node_history_response.json')),
     'svn': json.loads(fixture.load_resource('svn_node_history_response.json')),
 }
 
 
 
 @pytest.mark.usefixtures("app")
 class TestFilesController:
 
     def test_index(self, backend):
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name, revision='tip', f_path='/'))
         commit = backend.repo.get_commit()
 
         params = {
             'repo_name': backend.repo_name,
             'commit_id': commit.raw_id,
             'date': commit.date
         }
         assert_dirs_in_response(response, ['docs', 'vcs'], params)
         files = [
             '.gitignore',
             '.hgignore',
             '.hgtags',
             # TODO: missing in Git
             # '.travis.yml',
             'MANIFEST.in',
             'README.rst',
             # TODO: File is missing in svn repository
             # 'run_test_and_report.sh',
             'setup.cfg',
             'setup.py',
             'test_and_report.sh',
             'tox.ini',
         ]
         assert_files_in_response(response, files, params)
         assert_timeago_in_response(response, files, params)
 
     def test_index_links_submodules_with_absolute_url(self, backend_hg):
         repo = backend_hg['subrepos']
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=repo.repo_name, revision='tip', f_path='/'))
         assert_response = response.assert_response()
         assert_response.contains_one_link(
             'absolute-path @ 000000000000', 'http://example.com/absolute-path')
 
     def test_index_links_submodules_with_absolute_url_subpaths(
             self, backend_hg):
         repo = backend_hg['subrepos']
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=repo.repo_name, revision='tip', f_path='/'))
         assert_response = response.assert_response()
         assert_response.contains_one_link(
             'subpaths-path @ 000000000000',
             'http://sub-base.example.com/subpaths-path')
 
     @pytest.mark.xfail_backends("svn", reason="Depends on branch support")
     def test_files_menu(self, backend):
         new_branch = "temp_branch_name"
         commits = [
             {'message': 'a'},
             {'message': 'b', 'branch': new_branch}
         ]
         backend.create_repo(commits)
 
         backend.repo.landing_rev = "branch:%s" % new_branch
 
         # get response based on tip and not new revision
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name, revision='tip', f_path='/'),
             status=200)
 
         # make sure Files menu url is not tip but new revision
         landing_rev = backend.repo.landing_rev[1]
         files_url = url('files_home', repo_name=backend.repo_name,
                         revision=landing_rev)
 
         assert landing_rev != 'tip'
         response.mustcontain('<li class="active"><a class="menulink" href="%s">' % files_url)
 
     def test_index_commit(self, backend):
         commit = backend.repo.get_commit(commit_idx=32)
 
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name,
             revision=commit.raw_id,
             f_path='/')
         )
 
         dirs = ['docs', 'tests']
         files = ['README.rst']
         params = {
             'repo_name': backend.repo_name,
             'commit_id': commit.raw_id,
         }
         assert_dirs_in_response(response, dirs, params)
         assert_files_in_response(response, files, params)
 
     def test_index_different_branch(self, backend):
         branches = dict(
             hg=(150, ['git']),
             # TODO: Git test repository does not contain other branches
             git=(633, ['master']),
             # TODO: Branch support in Subversion
             svn=(150, [])
         )
         idx, branches = branches[backend.alias]
         commit = backend.repo.get_commit(commit_idx=idx)
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name,
             revision=commit.raw_id,
             f_path='/'))
         assert_response = response.assert_response()
         for branch in branches:
             assert_response.element_contains('.tags .branchtag', branch)
 
     def test_index_paging(self, backend):
         repo = backend.repo
         indexes = [73, 92, 109, 1, 0]
         idx_map = [(rev, repo.get_commit(commit_idx=rev).raw_id)
                    for rev in indexes]
 
         for idx in idx_map:
             response = self.app.get(url(
                 controller='files', action='index',
                 repo_name=backend.repo_name,
                 revision=idx[1],
                 f_path='/'))
 
             response.mustcontain("""r%s:%s""" % (idx[0], idx[1][:8]))
 
     def test_file_source(self, backend):
         commit = backend.repo.get_commit(commit_idx=167)
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name,
             revision=commit.raw_id,
             f_path='vcs/nodes.py'))
 
         msgbox = """<div class="commit right-content">%s</div>"""
         response.mustcontain(msgbox % (commit.message, ))
 
         assert_response = response.assert_response()
         if commit.branch:
             assert_response.element_contains('.tags.tags-main .branchtag', commit.branch)
         if commit.tags:
             for tag in commit.tags:
                 assert_response.element_contains('.tags.tags-main .tagtag', tag)
 
     def test_file_source_history(self, backend):
         response = self.app.get(
             url(
                 controller='files', action='history',
                 repo_name=backend.repo_name,
                 revision='tip',
                 f_path='vcs/nodes.py'),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert NODE_HISTORY[backend.alias] == json.loads(response.body)
 
     def test_file_source_history_svn(self, backend_svn):
         simple_repo = backend_svn['svn-simple-layout']
         response = self.app.get(
             url(
                 controller='files', action='history',
                 repo_name=simple_repo.repo_name,
                 revision='tip',
                 f_path='trunk/example.py'),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 
         expected_data = json.loads(
             fixture.load_resource('svn_node_history_branches.json'))
         assert expected_data == response.json
 
     def test_file_annotation_history(self, backend):
         response = self.app.get(
             url(
                 controller='files', action='history',
                 repo_name=backend.repo_name,
                 revision='tip',
                 f_path='vcs/nodes.py',
                 annotate=True),
             extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
         assert NODE_HISTORY[backend.alias] == json.loads(response.body)
 
     def test_file_annotation(self, backend):
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=backend.repo_name, revision='tip', f_path='vcs/nodes.py',
             annotate=True))
 
         expected_revisions = {
             'hg': 'r356',
             'git': 'r345',
             'svn': 'r208',
         }
         response.mustcontain(expected_revisions[backend.alias])
 
     def test_file_authors(self, backend):
         response = self.app.get(url(
             controller='files', action='authors',
             repo_name=backend.repo_name,
             revision='tip',
             f_path='vcs/nodes.py',
             annotate=True))
 
         expected_authors = {
             'hg': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'git': ('Marcin Kuzminski', 'Lukasz Balcerzak'),
             'svn': ('marcin', 'lukasz'),
         }
 
         for author in expected_authors[backend.alias]:
             response.mustcontain(author)
 
     def test_tree_search_top_level(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 revision=commit.raw_id, f_path='/'),
             extra_environ=xhr_header)
         assert 'nodes' in response.json
         assert {'name': 'docs', 'type': 'dir'} in response.json['nodes']
 
     def test_tree_search_at_path(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 revision=commit.raw_id, f_path='/docs'),
             extra_environ=xhr_header)
         assert 'nodes' in response.json
         nodes = response.json['nodes']
         assert {'name': 'docs/api', 'type': 'dir'} in nodes
         assert {'name': 'docs/index.rst', 'type': 'file'} in nodes
 
     def test_tree_search_at_path_missing_xhr(self, backend):
         self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 revision='tip', f_path=''), status=400)
 
     def test_tree_view_list(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 f_path='/', revision=commit.raw_id),
             extra_environ=xhr_header,
         )
         response.mustcontain("vcs/web/simplevcs/views/repository.py")
 
     def test_tree_view_list_at_path(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 f_path='/docs', revision=commit.raw_id),
             extra_environ=xhr_header,
         )
         response.mustcontain("docs/index.rst")
 
     def test_tree_view_list_missing_xhr(self, backend):
         self.app.get(
             url('files_nodelist_home', repo_name=backend.repo_name,
                 f_path='/', revision='tip'), status=400)
 
     def test_nodetree_full_success(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodetree_full', repo_name=backend.repo_name,
                 f_path='/', commit_id=commit.raw_id),
             extra_environ=xhr_header)
 
         assert_response = response.assert_response()
 
         for attr in ['data-commit-id', 'data-date', 'data-author']:
             elements = assert_response.get_elements('[{}]'.format(attr))
             assert len(elements) > 1
 
             for element in elements:
                 assert element.get(attr)
 
     def test_nodetree_full_if_file(self, backend, xhr_header):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(
             url('files_nodetree_full', repo_name=backend.repo_name,
                 f_path='README.rst', commit_id=commit.raw_id),
             extra_environ=xhr_header)
         assert response.body == ''
 
     def test_tree_metadata_list_missing_xhr(self, backend):
         self.app.get(
             url('files_nodetree_full', repo_name=backend.repo_name,
                 f_path='/', commit_id='tip'), status=400)
 
     def test_access_empty_repo_redirect_to_summary_with_alert_write_perms(
             self, app, backend_stub, autologin_regular_user, user_regular,
             user_util):
         repo = backend_stub.create_repo()
         user_util.grant_user_permission_to_repo(
             repo, user_regular, 'repository.write')
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=repo.repo_name, revision='tip', f_path='/'))
         assert_session_flash(
             response,
             'There are no files yet. <a class="alert-link" '
             'href="/%s/add/0/#edit">Click here to add a new file.</a>'
             % (repo.repo_name))
 
     def test_access_empty_repo_redirect_to_summary_with_alert_no_write_perms(
             self, backend_stub, user_util):
         repo = backend_stub.create_repo()
         repo_file_url = url(
             'files_add_home',
             repo_name=repo.repo_name,
             revision=0, f_path='', anchor='edit')
         response = self.app.get(url(
             controller='files', action='index',
             repo_name=repo.repo_name, revision='tip', f_path='/'))
         assert_not_in_session_flash(response, repo_file_url)
 
 
 # TODO: johbo: Think about a better place for these tests. Either controller
 # specific unit tests or we move down the whole logic further towards the vcs
 # layer
 class TestAdjustFilePathForSvn(object):
     """SVN specific adjustments of node history in FileController."""
 
     def test_returns_path_relative_to_matched_reference(self):
         repo = self._repo(branches=['trunk'])
         self.assert_file_adjustment('trunk/file', 'file', repo)
 
     def test_does_not_modify_file_if_no_reference_matches(self):
         repo = self._repo(branches=['trunk'])
         self.assert_file_adjustment('notes/file', 'notes/file', repo)
 
     def test_does_not_adjust_partial_directory_names(self):
         repo = self._repo(branches=['trun'])
         self.assert_file_adjustment('trunk/file', 'trunk/file', repo)
 
     def test_is_robust_to_patterns_which_prefix_other_patterns(self):
         repo = self._repo(branches=['trunk', 'trunk/new', 'trunk/old'])
         self.assert_file_adjustment('trunk/new/file', 'file', repo)
 
     def assert_file_adjustment(self, f_path, expected, repo):
         controller = FilesController()
         result = controller._adjust_file_path_for_svn(f_path, repo)
         assert result == expected
 
     def _repo(self, branches=None):
         repo = mock.Mock()
         repo.branches = OrderedDict((name, '0') for name in branches or [])
         repo.tags = {}
         return repo
 
 
 @pytest.mark.usefixtures("app")
 class TestRepositoryArchival(object):
 
     def test_archival(self, backend):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=173)
         for archive, info in settings.ARCHIVE_SPECS.items():
             mime_type, arch_ext = info
             short = commit.short_id + arch_ext
             fname = commit.raw_id + arch_ext
             filename = '%s-%s' % (backend.repo_name, short)
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=backend.repo_name,
                                         fname=fname))
 
             assert response.status == '200 OK'
             headers = {
                 'Pragma': 'no-cache',
                 'Cache-Control': 'no-cache',
                 'Content-Disposition': 'attachment; filename=%s' % filename,
                 'Content-Type': '%s; charset=utf-8' % mime_type,
             }
             if 'Set-Cookie' in response.response.headers:
                 del response.response.headers['Set-Cookie']
             assert response.response.headers == headers
 
     def test_archival_wrong_ext(self, backend):
         backend.enable_downloads()
         commit = backend.repo.get_commit(commit_idx=173)
         for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
             fname = commit.raw_id + arch_ext
 
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=backend.repo_name,
                                         fname=fname))
             response.mustcontain('Unknown archive type')
 
     def test_archival_wrong_commit_id(self, backend):
         backend.enable_downloads()
         for commit_id in ['00x000000', 'tar', 'wrong', '@##$@$42413232',
                           '232dffcd']:
             fname = '%s.zip' % commit_id
 
             response = self.app.get(url(controller='files',
                                         action='archivefile',
                                         repo_name=backend.repo_name,
                                         fname=fname))
             response.mustcontain('Unknown revision')
 
 
 @pytest.mark.usefixtures("app", "autologin_user")
 class TestRawFileHandling(object):
 
     def test_raw_file_ok(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=backend.repo_name,
                                     revision=commit.raw_id,
                                     f_path='vcs/nodes.py'))
 
         assert response.content_disposition == "attachment; filename=nodes.py"
         assert response.content_type == "text/x-python"
 
     def test_raw_file_wrong_cs(self, backend):
         commit_id = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
 
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=backend.repo_name,
                                     revision=commit_id,
                                     f_path=f_path), status=404)
 
         msg = """No such commit exists for this repository"""
         response.mustcontain(msg)
 
     def test_raw_file_wrong_f_path(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='rawfile',
                                     repo_name=backend.repo_name,
                                     revision=commit.raw_id,
                                     f_path=f_path), status=404)
 
         msg = (
             "There is no file nor directory at the given path: "
             "&#39;%s&#39; at commit %s" % (f_path, commit.short_id))
         response.mustcontain(msg)
 
     def test_raw_ok(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=backend.repo_name,
                                     revision=commit.raw_id,
                                     f_path='vcs/nodes.py'))
 
         assert response.content_type == "text/plain"
 
     def test_raw_wrong_cs(self, backend):
         commit_id = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
         f_path = 'vcs/nodes.py'
 
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=backend.repo_name,
                                     revision=commit_id,
                                     f_path=f_path), status=404)
 
         msg = """No such commit exists for this repository"""
         response.mustcontain(msg)
 
     def test_raw_wrong_f_path(self, backend):
         commit = backend.repo.get_commit(commit_idx=173)
         f_path = 'vcs/ERRORnodes.py'
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=backend.repo_name,
                                     revision=commit.raw_id,
                                     f_path=f_path), status=404)
         msg = (
             "There is no file nor directory at the given path: "
             "&#39;%s&#39; at commit %s" % (f_path, commit.short_id))
         response.mustcontain(msg)
 
     def test_raw_svg_should_not_be_rendered(self, backend):
         backend.create_repo()
         backend.ensure_file("xss.svg")
         response = self.app.get(url(controller='files', action='raw',
                                     repo_name=backend.repo_name,
                                     revision='tip',
                                     f_path='xss.svg'))
 
         # If the content type is image/svg+xml then it allows to render HTML
         # and malicious SVG.
         assert response.content_type == "text/plain"
 
 
 @pytest.mark.usefixtures("app")
 class TestFilesDiff:
 
     @pytest.mark.parametrize("diff", ['diff', 'download', 'raw'])
     def test_file_full_diff(self, backend, diff):
         commit1 = backend.repo.get_commit(commit_idx=-1)
         commit2 = backend.repo.get_commit(commit_idx=-2)
 
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=backend.repo_name,
                 f_path='README'),
             params={
                 'diff1': commit2.raw_id,
                 'diff2': commit1.raw_id,
                 'fulldiff': '1',
                 'diff': diff,
             })
 
         if diff == 'diff':
             # use redirect since this is OLD view redirecting to compare page
             response = response.follow()
 
         # It's a symlink to README.rst
         response.mustcontain('README.rst')
         response.mustcontain('No newline at end of file')
 
     def test_file_binary_diff(self, backend):
         commits = [
             {'message': 'First commit'},
             {'message': 'Commit with binary',
              'added': [nodes.FileNode('file.bin', content='\0BINARY\0')]},
         ]
         repo = backend.create_repo(commits=commits)
 
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=backend.repo_name,
                 f_path='file.bin'),
             params={
                 'diff1': repo.get_commit(commit_idx=0).raw_id,
                 'diff2': repo.get_commit(commit_idx=1).raw_id,
                 'fulldiff': '1',
                 'diff': 'diff',
             })
         # use redirect since this is OLD view redirecting to compare page
         response = response.follow()
         response.mustcontain('Expand 1 commit')
         response.mustcontain('1 file changed: 0 inserted, 0 deleted')
 
         if backend.alias == 'svn':
             response.mustcontain('new file 10644')
             # TODO(marcink): SVN doesn't yet detect binary changes
         else:
             response.mustcontain('new file 100644')
             response.mustcontain('binary diff hidden')
 
     def test_diff_2way(self, backend):
         commit1 = backend.repo.get_commit(commit_idx=-1)
         commit2 = backend.repo.get_commit(commit_idx=-2)
         response = self.app.get(
             url(
                 controller='files',
                 action='diff_2way',
                 repo_name=backend.repo_name,
                 f_path='README'),
             params={
                 'diff1': commit2.raw_id,
                 'diff2': commit1.raw_id,
             })
         # use redirect since this is OLD view redirecting to compare page
         response = response.follow()
 
         # It's a symlink to README.rst
         response.mustcontain('README.rst')
         response.mustcontain('No newline at end of file')
 
     def test_requires_one_commit_id(self, backend, autologin_user):
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=backend.repo_name,
                 f_path='README.rst'),
             status=400)
         response.mustcontain(
             'Need query parameter', 'diff1', 'diff2', 'to generate a diff.')
 
     def test_returns_no_files_if_file_does_not_exist(self, vcsbackend):
         repo = vcsbackend.repo
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=repo.name,
                 f_path='does-not-exist-in-any-commit',
                 diff1=repo[0].raw_id,
                 diff2=repo[1].raw_id),)
 
         response = response.follow()
         response.mustcontain('No files')
 
     def test_returns_redirect_if_file_not_changed(self, backend):
         commit = backend.repo.get_commit(commit_idx=-1)
         f_path = 'README'
         response = self.app.get(
             url(
                 controller='files',
                 action='diff_2way',
                 repo_name=backend.repo_name,
                 f_path=f_path,
                 diff1=commit.raw_id,
                 diff2=commit.raw_id,
             ),
         )
         response = response.follow()
         response.mustcontain('No files')
         response.mustcontain('No commits in this compare')
 
     def test_supports_diff_to_different_path_svn(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id_1 = '24'
         commit_id_2 = '26'
 
 
         print(            url(
                         controller='files',
                         action='diff',
                         repo_name=repo.name,
                         f_path='trunk/example.py',
                         diff1='tags/v0.2/example.py@' + commit_id_1,
                         diff2=commit_id_2))
 
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=repo.name,
                 f_path='trunk/example.py',
                 diff1='tags/v0.2/example.py@' + commit_id_1,
                 diff2=commit_id_2))
 
         response = response.follow()
         response.mustcontain(
             # diff contains this
             "Will print out a useful message on invocation.")
 
         # Note: Expecting that we indicate the user what's being compared
         response.mustcontain("trunk/example.py")
         response.mustcontain("tags/v0.2/example.py")
 
     def test_show_rev_redirects_to_svn_path(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id = repo[-1].raw_id
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=repo.name,
                 f_path='trunk/example.py',
                 diff1='branches/argparse/example.py@' + commit_id,
                 diff2=commit_id),
             params={'show_rev': 'Show at Revision'},
             status=302)
         assert response.headers['Location'].endswith(
             'svn-svn-simple-layout/files/26/branches/argparse/example.py')
 
     def test_show_rev_and_annotate_redirects_to_svn_path(self, backend_svn):
         #TODO: check this case
         return
 
         repo = backend_svn['svn-simple-layout'].scm_instance()
         commit_id = repo[-1].raw_id
         response = self.app.get(
             url(
                 controller='files',
                 action='diff',
                 repo_name=repo.name,
                 f_path='trunk/example.py',
                 diff1='branches/argparse/example.py@' + commit_id,
                 diff2=commit_id),
             params={
                 'show_rev': 'Show at Revision',
                 'annotate': 'true',
             },
             status=302)
         assert response.headers['Location'].endswith(
             'svn-svn-simple-layout/annotate/26/branches/argparse/example.py')
 
 
 @pytest.mark.usefixtures("app", "autologin_user")
 class TestChangingFiles:
 
     def test_add_file_view(self, backend):
         self.app.get(url(
             'files_add_home',
             repo_name=backend.repo_name,
             revision='tip', f_path='/'))
 
     @pytest.mark.xfail_backends("svn", reason="Depends on online editing")
     def test_add_file_into_repo_missing_content(self, backend, csrf_token):
         repo = backend.create_repo()
         filename = 'init.py'
         response = self.app.post(
             url(
                 'files_add',
                 repo_name=repo.repo_name,
                 revision='tip', f_path='/'),
             params={
                 'content': "",
                 'filename': filename,
                 'location': "",
                 'csrf_token': csrf_token,
             },
             status=302)
-        assert_session_flash(
-            response, 'Successfully committed to %s'
-            % os.path.join(filename))
+        assert_session_flash(response,
+            'Successfully committed new file `{}`'.format(os.path.join(filename)))
 
     def test_add_file_into_repo_missing_filename(self, backend, csrf_token):
         response = self.app.post(
             url(
                 'files_add',
                 repo_name=backend.repo_name,
                 revision='tip', f_path='/'),
             params={
                 'content': "foo",
                 'csrf_token': csrf_token,
             },
             status=302)
 
         assert_session_flash(response, 'No filename')
 
     def test_add_file_into_repo_errors_and_no_commits(
             self, backend, csrf_token):
         repo = backend.create_repo()
         # Create a file with no filename, it will display an error but
         # the repo has no commits yet
         response = self.app.post(
             url(
                 'files_add',
                 repo_name=repo.repo_name,
                 revision='tip', f_path='/'),
             params={
                 'content': "foo",
                 'csrf_token': csrf_token,
             },
             status=302)
 
         assert_session_flash(response, 'No filename')
 
         # Not allowed, redirect to the summary
         redirected = response.follow()
         summary_url = h.route_path('repo_summary', repo_name=repo.repo_name)
 
         # As there are no commits, displays the summary page with the error of
         # creating a file with no filename
 
         assert redirected.request.path == summary_url
 
     @pytest.mark.parametrize("location, filename", [
         ('/abs', 'foo'),
         ('../rel', 'foo'),
         ('file/../foo', 'foo'),
     ])
     def test_add_file_into_repo_bad_filenames(
             self, location, filename, backend, csrf_token):
         response = self.app.post(
             url(
                 'files_add',
                 repo_name=backend.repo_name,
                 revision='tip', f_path='/'),
             params={
                 'content': "foo",
                 'filename': filename,
                 'location': location,
                 'csrf_token': csrf_token,
             },
             status=302)
 
         assert_session_flash(
             response,
             'The location specified must be a relative path and must not '
             'contain .. in the path')
 
     @pytest.mark.parametrize("cnt, location, filename", [
         (1, '', 'foo.txt'),
         (2, 'dir', 'foo.rst'),
         (3, 'rel/dir', 'foo.bar'),
     ])
     def test_add_file_into_repo(self, cnt, location, filename, backend,
                                 csrf_token):
         repo = backend.create_repo()
         response = self.app.post(
             url(
                 'files_add',
                 repo_name=repo.repo_name,
                 revision='tip', f_path='/'),
             params={
                 'content': "foo",
                 'filename': filename,
                 'location': location,
                 'csrf_token': csrf_token,
             },
             status=302)
-        assert_session_flash(
-            response, 'Successfully committed to %s'
-            % os.path.join(location, filename))
+        assert_session_flash(response,
+            'Successfully committed new file `{}`'.format(
+                os.path.join(location, filename)))
 
     def test_edit_file_view(self, backend):
         response = self.app.get(
             url(
                 'files_edit_home',
                 repo_name=backend.repo_name,
                 revision=backend.default_head_id,
                 f_path='vcs/nodes.py'),
             status=200)
         response.mustcontain("Module holding everything related to vcs nodes.")
 
     def test_edit_file_view_not_on_branch(self, backend):
         repo = backend.create_repo()
         backend.ensure_file("vcs/nodes.py")
 
         response = self.app.get(
             url(
                 'files_edit_home',
                 repo_name=repo.repo_name,
                 revision='tip', f_path='vcs/nodes.py'),
             status=302)
         assert_session_flash(
             response,
             'You can only edit files with revision being a valid branch')
 
     def test_edit_file_view_commit_changes(self, backend, csrf_token):
         repo = backend.create_repo()
         backend.ensure_file("vcs/nodes.py", content="print 'hello'")
 
         response = self.app.post(
             url(
                 'files_edit',
                 repo_name=repo.repo_name,
                 revision=backend.default_head_id,
                 f_path='vcs/nodes.py'),
             params={
                 'content': "print 'hello world'",
                 'message': 'I committed',
                 'filename': "vcs/nodes.py",
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
-            response, 'Successfully committed to vcs/nodes.py')
+            response, 'Successfully committed changes to file `vcs/nodes.py`')
         tip = repo.get_commit(commit_idx=-1)
         assert tip.message == 'I committed'
 
     def test_edit_file_view_commit_changes_default_message(self, backend,
                                                            csrf_token):
         repo = backend.create_repo()
         backend.ensure_file("vcs/nodes.py", content="print 'hello'")
 
         commit_id = (
             backend.default_branch_name or
             backend.repo.scm_instance().commit_ids[-1])
 
         response = self.app.post(
             url(
                 'files_edit',
                 repo_name=repo.repo_name,
                 revision=commit_id,
                 f_path='vcs/nodes.py'),
             params={
                 'content': "print 'hello world'",
                 'message': '',
                 'filename': "vcs/nodes.py",
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
-            response, 'Successfully committed to vcs/nodes.py')
+            response, 'Successfully committed changes to file `vcs/nodes.py`')
         tip = repo.get_commit(commit_idx=-1)
         assert tip.message == 'Edited file vcs/nodes.py via RhodeCode Enterprise'
 
     def test_delete_file_view(self, backend):
         self.app.get(url(
             'files_delete_home',
             repo_name=backend.repo_name,
             revision='tip', f_path='vcs/nodes.py'))
 
     def test_delete_file_view_not_on_branch(self, backend):
         repo = backend.create_repo()
         backend.ensure_file('vcs/nodes.py')
 
         response = self.app.get(
             url(
                 'files_delete_home',
                 repo_name=repo.repo_name,
                 revision='tip', f_path='vcs/nodes.py'),
             status=302)
         assert_session_flash(
             response,
             'You can only delete files with revision being a valid branch')
 
     def test_delete_file_view_commit_changes(self, backend, csrf_token):
         repo = backend.create_repo()
         backend.ensure_file("vcs/nodes.py")
 
         response = self.app.post(
             url(
                 'files_delete_home',
                 repo_name=repo.repo_name,
                 revision=backend.default_head_id,
                 f_path='vcs/nodes.py'),
             params={
                 'message': 'i commited',
                 'csrf_token': csrf_token,
             },
             status=302)
         assert_session_flash(
-            response, 'Successfully deleted file vcs/nodes.py')
+            response, 'Successfully deleted file `vcs/nodes.py`')
 
 
 def assert_files_in_response(response, files, params):
     template = (
         'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
     _assert_items_in_response(response, files, template, params)
 
 
 def assert_dirs_in_response(response, dirs, params):
     template = (
         'href="/%(repo_name)s/files/%(commit_id)s/%(name)s"')
     _assert_items_in_response(response, dirs, template, params)
 
 
 def _assert_items_in_response(response, items, template, params):
     for item in items:
         item_params = {'name': item}
         item_params.update(params)
         response.mustcontain(template % item_params)
 
 
 def assert_timeago_in_response(response, items, params):
     for item in items:
         response.mustcontain(h.age_component(params['date']))