kallithea Changeset - 287b5f1cb40a

Changeset - 287b5f1cb40a

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Mads Kiilerich - 6 years ago 2019-12-16 00:02:34
mads@kiilerich.com

Grafted from: 26b6d73775ab

py3: only use safe_str for string conversion - not for arbitrary __str__ invocation

Usually, we actually just want unicode string, and it is better to just unicode().

6 files changed with 12 insertions and 13 deletions:

kallithea/controllers/api/__init__.py

kallithea/controllers/changelog.py

kallithea/controllers/files.py

kallithea/controllers/summary.py

kallithea/lib/base.py

kallithea/lib/vcs/utils/__init__.py

0 comments (0 inline, 0 general)

kallithea/controllers/api/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~
 JSON RPC controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import inspect
 import itertools
 import logging
 import time
 import traceback
 import types
 from tg import Response, TGController, request, response
 from webob.exc import HTTPError, HTTPException
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.base import _get_access_path
 from kallithea.lib.base import _get_ip_addr as _get_ip
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import User
 log = logging.getLogger('JSONRPC')
 class JSONRPCError(BaseException):
     def __init__(self, message):
         self.message = message
         super(JSONRPCError, self).__init__()
     def __str__(self):
         return safe_str(self.message)
 class JSONRPCErrorResponse(Response, HTTPException):
     """
     Generate a Response object with a JSON-RPC error body
     """
     def __init__(self, message=None, retid=None, code=None):
         HTTPException.__init__(self, message, self)
         Response.__init__(self,
                           json_body=dict(id=retid, result=None, error=message),
                           status=code,
                           content_type='application/json')
 class JSONRPCController(TGController):
     """
      A WSGI-speaking JSON-RPC controller class
      See the specification:
      <http://json-rpc.org/wiki/specification>`.
      Valid controller return values should be json-serializable objects.
      Sub-classes should catch their exceptions and raise JSONRPCError
      if they want to pass meaningful errors to the client.
      """
     def _get_ip_addr(self, environ):
         return _get_ip(environ)
     def _get_method_args(self):
         """
         Return `self._rpc_args` to dispatched controller method
         chosen by __call__
         """
         return self._rpc_args
     def _dispatch(self, state, remainder=None):
         """
         Parse the request body as JSON, look up the method on the
         controller and if it exists, dispatch to it.
         """
         # Since we are here we should respond as JSON
         response.content_type = 'application/json'
         environ = state.request.environ
         start = time.time()
         ip_addr = self._get_ip_addr(environ)
         self._req_id = None
         if 'CONTENT_LENGTH' not in environ:
             log.debug("No Content-Length")
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="No Content-Length in request")
         else:
             length = environ['CONTENT_LENGTH'] or 0
             length = int(environ['CONTENT_LENGTH'])
             log.debug('Content-Length: %s', length)
         if length == 0:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="Content-Length is 0")
         raw_body = environ['wsgi.input'].read(length)
         try:
             json_body = json.loads(raw_body)
         except ValueError as e:
             # catch JSON errors Here
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message="JSON parse error ERR:%s RAW:%r"
                                                 % (e, raw_body))
         # check AUTH based on API key
         try:
             self._req_api_key = json_body['api_key']
             self._req_id = json_body['id']
             self._req_method = json_body['method']
             self._request_params = json_body['args']
             if not isinstance(self._request_params, dict):
                 self._request_params = {}
             log.debug('method: %s, params: %s',
                       self._req_method, self._request_params)
         except KeyError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Incorrect JSON query missing %s' % e)
         # check if we can find this session using api_key
         try:
             u = User.get_by_api_key(self._req_api_key)
             auth_user = AuthUser.make(dbuser=u, ip_addr=ip_addr)
             if auth_user is None:
                 raise JSONRPCErrorResponse(retid=self._req_id,
                                            message='Invalid API key')
         except Exception as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message='Invalid API key')
         request.authuser = auth_user
         request.ip_addr = ip_addr
         self._error = None
         try:
             self._func = self._find_method()
         except AttributeError as e:
             raise JSONRPCErrorResponse(retid=self._req_id,
                                        message=str(e))
         # now that we have a method, add self._req_params to
         # self.kargs and dispatch control to WGIController
         argspec = inspect.getargspec(self._func)
         arglist = argspec[0][1:]
         defaults = [type(arg) for arg in argspec[3] or []]
         default_empty = type(NotImplemented)
         # kw arguments required by this method
         func_kwargs = dict(itertools.izip_longest(reversed(arglist), reversed(defaults),
                                                   fillvalue=default_empty))
         # This attribute will need to be first param of a method that uses
         # api_key, which is translated to instance of user at that name
         USER_SESSION_ATTR = 'apiuser'
         # get our arglist and check if we provided them as args
         for arg, default in func_kwargs.iteritems():
             if arg == USER_SESSION_ATTR:
                 # USER_SESSION_ATTR is something translated from API key and
                 # this is checked before so we don't need validate it
                 continue
             # skip the required param check if it's default value is
             # NotImplementedType (default_empty)
             if default == default_empty and arg not in self._request_params:
                 raise JSONRPCErrorResponse(
                     retid=self._req_id,
                     message='Missing non optional `%s` arg in JSON DATA' % arg,
+                )
         extra = set(self._request_params).difference(func_kwargs)
         if extra:
             raise JSONRPCErrorResponse(
                 retid=self._req_id,
                 message='Unknown %s arg in JSON DATA' %
                         ', '.join('`%s`' % arg for arg in extra),
+            )
         self._rpc_args = {}
         self._rpc_args.update(self._request_params)
         self._rpc_args['action'] = self._req_method
         self._rpc_args['environ'] = environ
         log.info('IP: %s Request to %s time: %.3fs' % (
             self._get_ip_addr(environ),
             safe_unicode(_get_access_path(environ)), time.time() - start)
+        )
         state.set_action(self._rpc_call, [])
         state.set_params(self._rpc_args)
         return state
     def _rpc_call(self, action, environ, **rpc_args):
         """
         Call the specified RPC Method
         """
         raw_response = ''
         try:
             raw_response = getattr(self, action)(**rpc_args)
             if isinstance(raw_response, HTTPError):
                 self._error = str(raw_response)
         except JSONRPCError as e:
-            self._error = safe_str(e)
+            self._error = unicode(e)
         except Exception as e:
             log.error('Encountered unhandled exception: %s',
                       traceback.format_exc(),)
             json_exc = JSONRPCError('Internal server error')
-            self._error = safe_str(json_exc)
+            self._error = unicode(json_exc)
         if self._error is not None:
             raw_response = None
         response = dict(id=self._req_id, result=raw_response, error=self._error)
         try:
             return json.dumps(response)
         except TypeError as e:
             log.error('API FAILED. Error encoding response: %s', e)
             return json.dumps(
                 dict(
                     id=self._req_id,
                     result=None,
                     error="Error encoding response"
+                )
+            )
     def _find_method(self):
         """
         Return method named by `self._req_method` in controller if able
         """
         log.debug('Trying to find JSON-RPC method: %s', self._req_method)
         if self._req_method.startswith('_'):
             raise AttributeError("Method not allowed")
         try:
             func = getattr(self, self._req_method, None)
         except UnicodeEncodeError:
             raise AttributeError("Problem decoding unicode in requested "
                                  "method name.")
         if isinstance(func, types.MethodType):
             return func
         else:
             raise AttributeError("No such method: %s" % (self._req_method,))

kallithea/controllers/changelog.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.changelog
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 changelog controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from tg import request, session
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPFound, HTTPNotFound
 import kallithea.lib.helpers as h
 from kallithea.config.routing import url
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, NodeDoesNotExistError, RepositoryError
 log = logging.getLogger(__name__)
 class ChangelogController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(ChangelogController, self)._before(*args, **kwargs)
         c.affected_files_cut_off = 60
     @staticmethod
     def __get_cs(rev, repo):
         """
         Safe way to get changeset. If error occur fail with error message.
         :param rev: revision to fetch
         :param repo: repo instance
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
             h.flash(_('There are no changesets yet'), category='error')
         except RepositoryError as e:
             log.error(traceback.format_exc())
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
         raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name, revision=None, f_path=None):
         limit = 2000
         default = 100
         if request.GET.get('size'):
             c.size = max(min(safe_int(request.GET.get('size')), limit), 1)
             session['changelog_size'] = c.size
             session.save()
         else:
             c.size = int(session.get('changelog_size', default))
         # min size must be 1
         c.size = max(c.size, 1)
         p = safe_int(request.GET.get('page'), 1)
         branch_name = request.GET.get('branch', None)
         if (branch_name and
             branch_name not in c.db_repo_scm_instance.branches and
             branch_name not in c.db_repo_scm_instance.closed_branches and
             not revision
         ):
             raise HTTPFound(location=url('changelog_file_home', repo_name=c.repo_name,
                                     revision=branch_name, f_path=f_path or ''))
         if revision == 'tip':
             revision = None
         c.changelog_for_path = f_path
         try:
             if f_path:
                 log.debug('generating changelog for path %s', f_path)
                 # get the history for the file !
                 tip_cs = c.db_repo_scm_instance.get_changeset()
                 try:
                     collection = tip_cs.get_file_history(f_path)
                 except (NodeDoesNotExistError, ChangesetError):
                     # this node is not present at tip !
                     try:
                         cs = self.__get_cs(revision, repo_name)
                         collection = cs.get_file_history(f_path)
                     except RepositoryError as e:
-                        h.flash(safe_str(e), category='warning')
+                        h.flash(unicode(e), category='warning')
                         raise HTTPFound(location=h.url('changelog_home', repo_name=repo_name))
             else:
                 collection = c.db_repo_scm_instance.get_changesets(start=0, end=revision,
                                                         branch_name=branch_name, reverse=True)
             c.total_cs = len(collection)
             c.cs_pagination = Page(collection, page=p, item_count=c.total_cs, items_per_page=c.size,
                                    branch=branch_name)
             page_revisions = [x.raw_id for x in c.cs_pagination]
             c.cs_comments = c.db_repo.get_comments(page_revisions)
             c.cs_statuses = c.db_repo.statuses(page_revisions)
         except EmptyRepositoryError as e:
-            h.flash(safe_str(e), category='warning')
+            h.flash(unicode(e), category='warning')
             raise HTTPFound(location=url('summary_home', repo_name=c.repo_name))
         except (RepositoryError, ChangesetDoesNotExistError, Exception) as e:
             log.error(traceback.format_exc())
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
             raise HTTPFound(location=url('changelog_home', repo_name=c.repo_name))
         c.branch_name = branch_name
         c.branch_filters = [('', _('None'))] + \
             [(k, k) for k in c.db_repo_scm_instance.branches.keys()]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.branch_filters += [('-', '-')] + \
                 [(k, prefix + k) for k in c.db_repo_scm_instance.closed_branches.keys()]
         revs = []
         if not f_path:
             revs = [x.revision for x in c.cs_pagination]
         c.jsdata = graph_data(c.db_repo_scm_instance, revs)
         c.revision = revision # requested revision ref
         c.first_revision = c.cs_pagination[0] # pagination is never empty here!
         return render('changelog/changelog.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def changelog_details(self, cs):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             c.cs = c.db_repo_scm_instance.get_changeset(cs)
             return render('changelog/changelog_details.html')
         raise HTTPNotFound()

kallithea/controllers/files.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.files
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Files controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import posixpath
 import shutil
 import tempfile
 import traceback
 from collections import OrderedDict
 from tg import request, response
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 from kallithea.config.routing import url
 from kallithea.controllers.changeset import _context_url, _ignorews_url, anchor_url, get_ignore_ws, get_line_ctx
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.exceptions import NonRelativePathError
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import convert_line_endings, detect_mode, safe_int, safe_str, str2bool
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import (
     ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, ImproperArchiveTypeError, NodeAlreadyExistsError, NodeDoesNotExistError, NodeError, RepositoryError, VCSError)
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.model.db import Repository
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 class FilesController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(FilesController, self)._before(*args, **kwargs)
     def __get_cs(self, rev, silent_empty=False):
         """
         Safe way to get changeset if error occur it redirects to tip with
         proper message
         :param rev: revision to fetch
         :silent_empty: return None if repository is empty
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
             if silent_empty:
                 return None
             url_ = url('files_add_home',
                        repo_name=c.repo_name,
                        revision=0, f_path='', anchor='edit')
             add_new = h.link_to(_('Click here to add new file'), url_, class_="alert-link")
             h.flash(_('There are no files yet.') + ' ' + add_new, category='warning')
             raise HTTPNotFound()
         except (ChangesetDoesNotExistError, LookupError):
             msg = _('Such revision does not exist for this repository')
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
             raise HTTPNotFound()
     def __get_filenode(self, cs, path):
         """
         Returns file_node or raise HTTP error.
         :param cs: given changeset
         :param path: path to lookup
         """
         try:
             file_node = cs.get_node(path)
             if file_node.is_dir():
                 raise RepositoryError('given path is a directory')
         except ChangesetDoesNotExistError:
             msg = _('Such revision does not exist for this repository')
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
             raise HTTPNotFound()
         return file_node
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name, revision, f_path, annotate=False):
         # redirect to given revision from form if given
         post_revision = request.POST.get('at_rev', None)
         if post_revision:
             cs = self.__get_cs(post_revision) # FIXME - unused!
         c.revision = revision
         c.changeset = self.__get_cs(revision)
         c.branch = request.GET.get('branch', None)
         c.f_path = f_path
         c.annotate = annotate
         cur_rev = c.changeset.revision
         # used in files_source.html:
         c.cut_off_limit = self.cut_off_limit
         c.fulldiff = request.GET.get('fulldiff')
         # prev link
         try:
             prev_rev = c.db_repo_scm_instance.get_changeset(cur_rev).prev(c.branch)
             c.url_prev = url('files_home', repo_name=c.repo_name,
                          revision=prev_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_prev += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_prev = '#'
         # next link
         try:
             next_rev = c.db_repo_scm_instance.get_changeset(cur_rev).next(c.branch)
             c.url_next = url('files_home', repo_name=c.repo_name,
                      revision=next_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_next += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_next = '#'
         # files or dirs
         try:
             c.file = c.changeset.get_node(f_path)
             if c.file.is_submodule():
                 raise HTTPFound(location=c.file.url)
             elif c.file.is_file():
                 c.load_full_history = False
                 # determine if we're on branch head
                 _branches = c.db_repo_scm_instance.branches
                 c.on_branch_head = revision in _branches or revision in _branches.values()
                 _hist = []
                 c.file_history = []
                 if c.load_full_history:
                     c.file_history, _hist = self._get_node_history(c.changeset, f_path)
                 c.authors = []
                 for a in set([x.author for x in _hist]):
                     c.authors.append((h.email(a), h.person(a)))
             else:
                 c.authors = c.file_history = []
         except RepositoryError as e:
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
             raise HTTPNotFound()
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('files/files_ypjax.html')
         # TODO: tags and bookmarks?
         c.revision_options = [(c.changeset.raw_id,
                               _('%s at %s') % (b, h.short_id(c.changeset.raw_id))) for b in c.changeset.branches] + \
             [(n, b) for b, n in c.db_repo_scm_instance.branches.items()]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.revision_options += [('-', '-')] + \
                 [(n, prefix + b) for b, n in c.db_repo_scm_instance.closed_branches.items()]
         return render('files/files.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def history(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, 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(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def authors(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, f_path)
             c.authors = []
             for a in set([x.author for x in _hist]):
                 c.authors.append((h.email(a), h.person(a)))
             return render('files/files_history_box.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def rawfile(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, f_path)
         response.content_disposition = 'attachment; filename=%s' % \
             safe_str(f_path.split(Repository.url_sep())[-1])
         response.content_type = file_node.mimetype
         return file_node.content
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def raw(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, 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:
             'image/x-icon': ('image/x-icon', 'inline'),
             'image/png': ('image/png', 'inline'),
             'image/gif': ('image/gif', 'inline'),
             'image/jpeg': ('image/jpeg', 'inline'),
             'image/svg+xml': ('image/svg+xml', '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
         return file_node.content
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def delete(self, repo_name, revision, f_path):
         repo = c.db_repo
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches and revision not in _branches.values():
             h.flash(_('You can only delete files with revision '
                       'being a valid branch'), category='warning')
             raise HTTPFound(location=h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         c.default_message = _('Deleted file %s via Kallithea') % (f_path)
         c.f_path = f_path
         node_path = f_path
         author = request.authuser.full_contact
         if r_post:
             message = r_post.get('message') or c.default_message
             try:
                 nodes = {
                     node_path: {
                         'content': ''
+                    }
+                }
                 self.scm_model.delete_nodes(
                     user=request.authuser.user_id,
                     ip_addr=request.ip_addr,
                     repo=c.db_repo,
                     message=message,
                     nodes=nodes,
                     parent_cs=c.cs,
                     author=author,
+                )
                 h.flash(_('Successfully deleted file %s') % f_path,
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_delete.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def edit(self, repo_name, revision, f_path):
         repo = c.db_repo
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches and revision not in _branches.values():
             h.flash(_('You can only edit files with revision '
                       'being a valid branch'), category='warning')
             raise HTTPFound(location=h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         if c.file.is_binary:
             raise HTTPFound(location=url('files_home', repo_name=c.repo_name,
                             revision=c.cs.raw_id, f_path=f_path))
         c.default_message = _('Edited file %s via Kallithea') % (f_path)
         c.f_path = f_path
         if r_post:
             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(r_post.get('content', ''), mode)
             message = r_post.get('message') or c.default_message
             author = request.authuser.full_contact
             if content == old_content:
                 h.flash(_('No changes'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             try:
                 self.scm_model.commit_change(repo=c.db_repo_scm_instance,
                                              repo_name=repo_name, cs=c.cs,
                                              user=request.authuser.user_id,
                                              ip_addr=request.ip_addr,
                                              author=author, message=message,
                                              content=content, f_path=f_path)
                 h.flash(_('Successfully committed to %s') % f_path,
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_edit.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def add(self, repo_name, revision, f_path):
         repo = c.db_repo
         r_post = request.POST
         c.cs = self.__get_cs(revision, silent_empty=True)
         if c.cs is None:
             c.cs = EmptyChangeset(alias=c.db_repo_scm_instance.alias)
         c.default_message = (_('Added file via Kallithea'))
         c.f_path = f_path
         if r_post:
             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', '')
             file_obj = r_post.get('upload_file', None)
             if file_obj is not None and hasattr(file_obj, 'filename'):
                 filename = file_obj.filename
                 content = file_obj.file
                 if hasattr(content, 'file'):
                     # non posix systems store real file under file attr
                     content = content.file
             if not content:
                 h.flash(_('No content'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             if not filename:
                 h.flash(_('No filename'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             # strip all crap out of file, just leave the basename
             filename = os.path.basename(filename)
             node_path = posixpath.join(location, filename)
             author = request.authuser.full_contact
             try:
                 nodes = {
                     node_path: {
                         'content': content
+                    }
+                }
                 self.scm_model.create_nodes(
                     user=request.authuser.user_id,
                     ip_addr=request.ip_addr,
                     repo=c.db_repo,
                     message=message,
                     nodes=nodes,
                     parent_cs=c.cs,
                     author=author,
+                )
                 h.flash(_('Successfully committed to %s') % node_path,
                         category='success')
             except NonRelativePathError as e:
                 h.flash(_('Location must be relative path and must not '
                           'contain .. in path'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             except (NodeError, NodeAlreadyExistsError) as e:
                 h.flash(_(e), category='error')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_add.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def archivefile(self, repo_name, fname):
         fileformat = None
         revision = 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]
                 revision = archive_spec[0]
                 ext = ext_data[1]
         try:
             dbrepo = RepoModel().get_by_repo_name(repo_name)
             if not dbrepo.enable_downloads:
                 return _('Downloads disabled') # TODO: do something else?
             if c.db_repo_scm_instance.alias == 'hg':
                 # patch and reset hooks section of UI config to not run any
                 # hooks on fetching archives with subrepos
                 for k, v in c.db_repo_scm_instance._repo.ui.configitems('hooks'):
                     c.db_repo_scm_instance._repo.ui.setconfig('hooks', k, None)
             cs = c.db_repo_scm_instance.get_changeset(revision)
             content_type = settings.ARCHIVE_SPECS[fileformat][0]
         except ChangesetDoesNotExistError:
             return _('Unknown revision %s') % revision
         except EmptyRepositoryError:
             return _('Empty repository')
         except (ImproperArchiveTypeError, KeyError):
             return _('Unknown archive type')
         from kallithea import CONFIG
         rev_name = cs.raw_id[:12]
         archive_name = '%s-%s%s' % (safe_str(repo_name.replace('/', '_')),
                                     safe_str(rev_name), ext)
         archive_path = None
         cached_archive_path = None
         archive_cache_dir = CONFIG.get('archive_cache_dir')
         if archive_cache_dir and not subrepos: # TODO: subrepo caching?
             if not os.path.isdir(archive_cache_dir):
                 os.makedirs(archive_cache_dir)
             cached_archive_path = os.path.join(archive_cache_dir, archive_name)
             if os.path.isfile(cached_archive_path):
                 log.debug('Found cached archive in %s', cached_archive_path)
                 archive_path = cached_archive_path
             else:
                 log.debug('Archive %s is not yet cached', archive_name)
         if archive_path is None:
             # generate new archive
             fd, archive_path = tempfile.mkstemp()
             log.debug('Creating new temp archive in %s', archive_path)
             with os.fdopen(fd, 'wb') as stream:
                 cs.fill_archive(stream=stream, kind=fileformat, subrepos=subrepos)
                 # stream (and thus fd) has been closed by cs.fill_archive
             if cached_archive_path is not None:
                 # we generated the archive - move it to cache
                 log.debug('Storing new archive in %s', cached_archive_path)
                 shutil.move(archive_path, cached_archive_path)
                 archive_path = cached_archive_path
         def get_chunked_archive(archive_path):
             stream = open(archive_path, 'rb')
             while True:
                 data = stream.read(16 * 1024)
                 if not data:
                     break
                 yield data
             stream.close()
             if archive_path != cached_archive_path:
                 log.debug('Destroying temp archive %s', archive_path)
                 os.remove(archive_path)
         action_logger(user=request.authuser,
                       action='user_downloaded_archive:%s' % (archive_name),
                       repo=repo_name, ipaddr=request.ip_addr, commit=True)
         response.content_disposition = str('attachment; filename=%s' % (archive_name))
         response.content_type = str(content_type)
         return get_chunked_archive(archive_path)
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')

kallithea/controllers/summary.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # 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 General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.summary
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Summary controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 18, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import calendar
 import itertools
 import logging
 import traceback
 from datetime import date, timedelta
 from time import mktime
 from beaker.cache import cache_region
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest
 import kallithea.lib.helpers as h
 from kallithea.config.conf import ALL_EXTS, ALL_READMES, LANGUAGES_EXTENSIONS_MAP
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.celerylib.tasks import get_commits_stats
 from kallithea.lib.compat import json
 from kallithea.lib.markup_renderer import MarkupRenderer
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetError, EmptyRepositoryError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.model.db import Statistics
 log = logging.getLogger(__name__)
 README_FILES = [''.join([x[0][0], x[1][0]]) for x in
                     sorted(list(itertools.product(ALL_READMES, ALL_EXTS)),
                            key=lambda y:y[0][1] + y[1][1])]
 class SummaryController(BaseRepoController):
     def __get_readme_data(self, db_repo):
         repo_name = db_repo.repo_name
         log.debug('Looking for README file')
         @cache_region('long_term', '_get_readme_from_cache')
         def _get_readme_from_cache(*_cache_keys):  # parameters are not really used - only as caching key
             readme_data = None
             readme_file = None
             try:
                 # gets the landing revision! or tip if fails
                 cs = db_repo.get_landing_changeset()
                 if isinstance(cs, EmptyChangeset):
                     raise EmptyRepositoryError()
                 renderer = MarkupRenderer()
                 for f in README_FILES:
                     try:
                         readme = cs.get_node(f)
                         if not isinstance(readme, FileNode):
                             continue
                         readme_file = f
                         log.debug('Found README file `%s` rendering...',
                                   readme_file)
                         readme_data = renderer.render(readme.content,
                                                       filename=f)
                         break
                     except NodeDoesNotExistError:
                         continue
             except ChangesetError:
                 log.error(traceback.format_exc())
                 pass
             except EmptyRepositoryError:
                 pass
             return readme_data, readme_file
         kind = 'README'
         return _get_readme_from_cache(repo_name, kind, c.db_repo.changeset_cache.get('raw_id'))
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name):
         p = safe_int(request.GET.get('page'), 1)
         size = safe_int(request.GET.get('size'), 10)
         try:
             collection = c.db_repo_scm_instance.get_changesets(reverse=True)
         except EmptyRepositoryError as e:
-            h.flash(safe_str(e), category='warning')
+            h.flash(unicode(e), category='warning')
             collection = []
         c.cs_pagination = Page(collection, page=p, items_per_page=size)
         page_revisions = [x.raw_id for x in list(c.cs_pagination)]
         c.cs_comments = c.db_repo.get_comments(page_revisions)
         c.cs_statuses = c.db_repo.statuses(page_revisions)
         c.ssh_repo_url = None
         if request.authuser.is_default_user:
             username = None
         else:
             username = request.authuser.username
             if c.ssh_enabled:
                 c.ssh_repo_url = c.db_repo.clone_url(clone_uri_tmpl=c.clone_ssh_tmpl)
         c.clone_repo_url = c.db_repo.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=False, username=username)
         c.clone_repo_url_id = c.db_repo.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True, username=username)
         if c.db_repo.enable_statistics:
             c.show_stats = True
         else:
             c.show_stats = False
         stats = Statistics.query() \
             .filter(Statistics.repository == c.db_repo) \
             .scalar()
         c.stats_percentage = 0
         if stats and stats.languages:
             c.no_data = False is c.db_repo.enable_statistics
             lang_stats_d = json.loads(stats.languages)
             lang_stats = [(x, {"count": y,
                                "desc": LANGUAGES_EXTENSIONS_MAP.get(x, '?')})
                           for x, y in lang_stats_d.items()]
             lang_stats.sort(key=lambda k: (-k[1]['count'], k[0]))
             c.trending_languages = lang_stats[:10]
         else:
             c.no_data = True
             c.trending_languages = []
         c.enable_downloads = c.db_repo.enable_downloads
         c.readme_data, c.readme_file = \
             self.__get_readme_data(c.db_repo)
         return render('summary/summary.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def repo_size(self, repo_name):
         if request.is_xhr:
             return c.db_repo._repo_size()
         else:
             raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def statistics(self, repo_name):
         if c.db_repo.enable_statistics:
             c.show_stats = True
             c.no_data_msg = _('No data ready yet')
         else:
             c.show_stats = False
             c.no_data_msg = _('Statistics are disabled for this repository')
         td = date.today() + timedelta(days=1)
         td_1m = td - timedelta(days=calendar.mdays[td.month])
         td_1y = td - timedelta(days=365)
         ts_min_m = mktime(td_1m.timetuple())
         ts_min_y = mktime(td_1y.timetuple())
         ts_max_y = mktime(td.timetuple())
         c.ts_min = ts_min_m
         c.ts_max = ts_max_y
         stats = Statistics.query() \
             .filter(Statistics.repository == c.db_repo) \
             .scalar()
         c.stats_percentage = 0
         if stats and stats.languages:
             c.no_data = False is c.db_repo.enable_statistics
             lang_stats_d = json.loads(stats.languages)
             c.commit_data = json.loads(stats.commit_activity)
             c.overview_data = json.loads(stats.commit_activity_combined)
             lang_stats = ((x, {"count": y,
                                "desc": LANGUAGES_EXTENSIONS_MAP.get(x)})
                           for x, y in lang_stats_d.items())
             c.trending_languages = (
                 sorted(lang_stats, reverse=True, key=lambda k: k[1])[:10]
+            )
             last_rev = stats.stat_on_revision + 1
             c.repo_last_rev = c.db_repo_scm_instance.count() \
                 if c.db_repo_scm_instance.revisions else 0
             if last_rev == 0 or c.repo_last_rev == 0:
                 pass
             else:
                 c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                                 c.repo_last_rev) * 100)
         else:
             c.commit_data = {}
             c.overview_data = ([[ts_min_y, 0], [ts_max_y, 10]])
             c.trending_languages = {}
             c.no_data = True
         recurse_limit = 500  # don't recurse more than 500 times when parsing
         get_commits_stats(c.db_repo.repo_name, ts_min_y, ts_max_y, recurse_limit)
         return render('summary/statistics.html')

kallithea/lib/base.py

➞

Show inline comments

@@ @@ -222,423 +222,423 @@ class BaseVCSController(object): @@
         default_authuser = AuthUser.make(dbuser=default_user, ip_addr=ip_addr)
         if default_authuser is None:
             log.debug('No anonymous access at all') # move on to proper user auth
         else:
             if self._check_permission(action, default_authuser, repo_name):
                 return default_authuser, None
             log.debug('Not authorized to access this repository as anonymous user')
         username = None
         #==============================================================
         # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
         # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
         #==============================================================
         # try to auth based on environ, container auth methods
         log.debug('Running PRE-AUTH for container based authentication')
         pre_auth = auth_modules.authenticate('', '', environ)
         if pre_auth is not None and pre_auth.get('username'):
             username = pre_auth['username']
         log.debug('PRE-AUTH got %s as username', username)
         # If not authenticated by the container, running basic auth
         if not username:
             self.authenticate.realm = safe_str(self.config['realm'])
             result = self.authenticate(environ)
             if isinstance(result, str):
                 paste.httpheaders.AUTH_TYPE.update(environ, 'basic')
                 paste.httpheaders.REMOTE_USER.update(environ, result)
                 username = result
             else:
                 return None, result.wsgi_application
         #==============================================================
         # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
         #==============================================================
         try:
             user = User.get_by_username_or_email(username)
         except Exception:
             log.error(traceback.format_exc())
             return None, webob.exc.HTTPInternalServerError()
         authuser = AuthUser.make(dbuser=user, ip_addr=ip_addr)
         if authuser is None:
             return None, webob.exc.HTTPForbidden()
         if not self._check_permission(action, authuser, repo_name):
             return None, webob.exc.HTTPForbidden()
         return user, None
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     def _check_permission(self, action, authuser, repo_name):
         """
         Checks permissions using action (push/pull) user and repository
         name
         :param action: 'push' or 'pull' action
         :param user: `User` instance
         :param repo_name: repository name
         """
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
         else:
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
         return True
     def _get_ip_addr(self, environ):
         return _get_ip_addr(environ)
     def __call__(self, environ, start_response):
         try:
             # try parsing a request for this VCS - if it fails, call the wrapped app
             parsed_request = self.parse_request(environ)
             if parsed_request is None:
                 return self.application(environ, start_response)
             # skip passing error to error controller
             environ['pylons.status_code_redirect'] = True
             # quick check if repo exists...
             if not is_valid_repo(parsed_request.repo_name, self.basepath, self.scm_alias):
                 raise webob.exc.HTTPNotFound()
             if parsed_request.action is None:
                 # Note: the client doesn't get the helpful error message
                 raise webob.exc.HTTPBadRequest('Unable to detect pull/push action for %r! Are you using a nonstandard command or client?' % parsed_request.repo_name)
             #======================================================================
             # CHECK PERMISSIONS
             #======================================================================
             ip_addr = self._get_ip_addr(environ)
             user, response_app = self._authorize(environ, parsed_request.action, parsed_request.repo_name, ip_addr)
             if response_app is not None:
                 return response_app(environ, start_response)
             #======================================================================
             # REQUEST HANDLING
             #======================================================================
             set_hook_environment(user.username, ip_addr,
                 parsed_request.repo_name, self.scm_alias, parsed_request.action)
             try:
                 log.info('%s action on %s repo "%s" by "%s" from %s',
                          parsed_request.action, self.scm_alias, parsed_request.repo_name, safe_str(user.username), ip_addr)
                 app = self._make_app(parsed_request)
                 return app(environ, start_response)
             except Exception:
                 log.error(traceback.format_exc())
                 raise webob.exc.HTTPInternalServerError()
         except webob.exc.HTTPException as e:
             return e(environ, start_response)
 class BaseController(TGController):
     def _before(self, *args, **kwargs):
         """
         _before is called before controller methods and after __call__
         """
         if request.needs_csrf_check:
             # CSRF protection: Whenever a request has ambient authority (whether
             # through a session cookie or its origin IP address), it must include
             # the correct token, unless the HTTP method is GET or HEAD (and thus
             # guaranteed to be side effect free. In practice, the only situation
             # where we allow side effects without ambient authority is when the
             # authority comes from an API key; and that is handled above.
             from kallithea.lib import helpers as h
             token = request.POST.get(h.session_csrf_secret_name)
             if not token or token != h.session_csrf_secret_token():
                 log.error('CSRF check failed')
                 raise webob.exc.HTTPForbidden()
         c.kallithea_version = __version__
         rc_config = Setting.get_app_settings()
         # Visual options
         c.visual = AttributeDict({})
         ## DB stored
         c.visual.show_public_icon = str2bool(rc_config.get('show_public_icon'))
         c.visual.show_private_icon = str2bool(rc_config.get('show_private_icon'))
         c.visual.stylify_metalabels = str2bool(rc_config.get('stylify_metalabels'))
         c.visual.page_size = safe_int(rc_config.get('dashboard_items', 100))
         c.visual.admin_grid_items = safe_int(rc_config.get('admin_grid_items', 100))
         c.visual.repository_fields = str2bool(rc_config.get('repository_fields'))
         c.visual.show_version = str2bool(rc_config.get('show_version'))
         c.visual.use_gravatar = str2bool(rc_config.get('use_gravatar'))
         c.visual.gravatar_url = rc_config.get('gravatar_url')
         c.ga_code = rc_config.get('ga_code')
         # TODO: replace undocumented backwards compatibility hack with db upgrade and rename ga_code
         if c.ga_code and '<' not in c.ga_code:
             c.ga_code = '''<script type="text/javascript">
                 var _gaq = _gaq || [];
                 _gaq.push(['_setAccount', '%s']);
                 _gaq.push(['_trackPageview']);
                 (function() {
                     var ga = document.createElement('script'); ga.type = 'text/javascript'; ga.async = true;
                     ga.src = ('https:' == document.location.protocol ? 'https://ssl' : 'http://www') + '.google-analytics.com/ga.js';
                     var s = document.getElementsByTagName('script')[0]; s.parentNode.insertBefore(ga, s);
                     })();
             </script>''' % c.ga_code
         c.site_name = rc_config.get('title')
         c.clone_uri_tmpl = rc_config.get('clone_uri_tmpl') or Repository.DEFAULT_CLONE_URI
         c.clone_ssh_tmpl = rc_config.get('clone_ssh_tmpl') or Repository.DEFAULT_CLONE_SSH
         ## INI stored
         c.visual.allow_repo_location_change = str2bool(config.get('allow_repo_location_change', True))
         c.visual.allow_custom_hooks_settings = str2bool(config.get('allow_custom_hooks_settings', True))
         c.ssh_enabled = str2bool(config.get('ssh_enabled', False))
         c.instance_id = config.get('instance_id')
         c.issues_url = config.get('bugtracker', url('issues_url'))
         # END CONFIG VARS
         c.repo_name = get_repo_slug(request)  # can be empty
         c.backends = list(BACKENDS)
         self.cut_off_limit = safe_int(config.get('cut_off_limit'))
         c.my_pr_count = PullRequest.query(reviewer_id=request.authuser.user_id, include_closed=False).count()
         self.scm_model = ScmModel()
     @staticmethod
     def _determine_auth_user(session_authuser, ip_addr):
         """
         Create an `AuthUser` object given the API key/bearer token
         (if any) and the value of the authuser session cookie.
         Returns None if no valid user is found (like not active or no access for IP).
         """
         # Authenticate by session cookie
         # In ancient login sessions, 'authuser' may not be a dict.
         # In that case, the user will have to log in again.
         # v0.3 and earlier included an 'is_authenticated' key; if present,
         # this must be True.
         if isinstance(session_authuser, dict) and session_authuser.get('is_authenticated', True):
             return AuthUser.from_cookie(session_authuser, ip_addr=ip_addr)
         # Authenticate by auth_container plugin (if enabled)
         if any(
             plugin.is_container_auth
             for plugin in auth_modules.get_auth_plugins()
         ):
             try:
                 user_info = auth_modules.authenticate('', '', request.environ)
             except UserCreationError as e:
                 from kallithea.lib import helpers as h
                 h.flash(e, 'error', logf=log.error)
             else:
                 if user_info is not None:
                     username = user_info['username']
                     user = User.get_by_username(username, case_insensitive=True)
                     return log_in_user(user, remember=False, is_external_auth=True, ip_addr=ip_addr)
         # User is default user (if active) or anonymous
         default_user = User.get_default_user(cache=True)
         authuser = AuthUser.make(dbuser=default_user, ip_addr=ip_addr)
         if authuser is None: # fall back to anonymous
             authuser = AuthUser(dbuser=default_user) # TODO: somehow use .make?
         return authuser
     @staticmethod
     def _basic_security_checks():
         """Perform basic security/sanity checks before processing the request."""
         # Only allow the following HTTP request methods.
         if request.method not in ['GET', 'HEAD', 'POST']:
             raise webob.exc.HTTPMethodNotAllowed()
         # Also verify the _method override - no longer allowed.
         if request.params.get('_method') is None:
             pass # no override, no problem
         else:
             raise webob.exc.HTTPMethodNotAllowed()
         # Make sure CSRF token never appears in the URL. If so, invalidate it.
         from kallithea.lib import helpers as h
         if h.session_csrf_secret_name in request.GET:
             log.error('CSRF key leak detected')
             session.pop(h.session_csrf_secret_name, None)
             session.save()
             h.flash(_('CSRF token leak has been detected - all form tokens have been expired'),
                     category='error')
         # WebOb already ignores request payload parameters for anything other
         # than POST/PUT, but double-check since other Kallithea code relies on
         # this assumption.
         if request.method not in ['POST', 'PUT'] and request.POST:
             log.error('%r request with payload parameters; WebOb should have stopped this', request.method)
             raise webob.exc.HTTPBadRequest()
     def __call__(self, environ, context):
         try:
             ip_addr = _get_ip_addr(environ)
             self._basic_security_checks()
             api_key = request.GET.get('api_key')
             try:
                 # Request.authorization may raise ValueError on invalid input
                 type, params = request.authorization
             except (ValueError, TypeError):
                 pass
             else:
                 if type.lower() == 'bearer':
                     api_key = params # bearer token is an api key too
             if api_key is None:
                 authuser = self._determine_auth_user(
                     session.get('authuser'),
                     ip_addr=ip_addr,
+                )
                 needs_csrf_check = request.method not in ['GET', 'HEAD']
             else:
                 dbuser = User.get_by_api_key(api_key)
                 if dbuser is None:
                     log.info('No db user found for authentication with API key ****%s from %s',
                              api_key[-4:], ip_addr)
                 authuser = AuthUser.make(dbuser=dbuser, is_external_auth=True, ip_addr=ip_addr)
                 needs_csrf_check = False # API key provides CSRF protection
             if authuser is None:
                 log.info('No valid user found')
                 raise webob.exc.HTTPForbidden()
             # set globals for auth user
             request.authuser = authuser
             request.ip_addr = ip_addr
             request.needs_csrf_check = needs_csrf_check
             log.info('IP: %s User: %s accessed %s',
                 request.ip_addr, request.authuser,
                 safe_unicode(_get_access_path(environ)),
+            )
             return super(BaseController, self).__call__(environ, context)
         except webob.exc.HTTPException as e:
             return e
 class BaseRepoController(BaseController):
     """
     Base class for controllers responsible for loading all needed data for
     repository loaded items are
     c.db_repo_scm_instance: instance of scm repository
     c.db_repo: instance of db
     c.repository_followers: number of followers
     c.repository_forks: number of forks
     c.repository_following: weather the current user is following the current repo
     """
     def _before(self, *args, **kwargs):
         super(BaseRepoController, self)._before(*args, **kwargs)
         if c.repo_name:  # extracted from routes
             _dbr = Repository.get_by_repo_name(c.repo_name)
             if not _dbr:
                 return
             log.debug('Found repository in database %s with state `%s`',
                       safe_unicode(_dbr), safe_unicode(_dbr.repo_state))
             route = getattr(request.environ.get('routes.route'), 'name', '')
             # allow to delete repos that are somehow damages in filesystem
             if route in ['delete_repo']:
                 return
             if _dbr.repo_state in [Repository.STATE_PENDING]:
                 if route in ['repo_creating_home']:
                     return
                 check_url = url('repo_creating_home', repo_name=c.repo_name)
                 raise webob.exc.HTTPFound(location=check_url)
             dbr = c.db_repo = _dbr
             c.db_repo_scm_instance = c.db_repo.scm_instance
             if c.db_repo_scm_instance is None:
                 log.error('%s this repository is present in database but it '
                           'cannot be created as an scm instance', c.repo_name)
                 from kallithea.lib import helpers as h
                 h.flash(_('Repository not found in the filesystem'),
                         category='error')
                 raise webob.exc.HTTPNotFound()
             # some globals counter for menu
             c.repository_followers = self.scm_model.get_followers(dbr)
             c.repository_forks = self.scm_model.get_forks(dbr)
             c.repository_pull_requests = self.scm_model.get_pull_requests(dbr)
             c.repository_following = self.scm_model.is_following_repo(
                                     c.repo_name, request.authuser.user_id)
     @staticmethod
     def _get_ref_rev(repo, ref_type, ref_name, returnempty=False):
         """
         Safe way to get changeset. If error occurs show error.
         """
         from kallithea.lib import helpers as h
         try:
             return repo.scm_instance.get_ref_revision(ref_type, ref_name)
         except EmptyRepositoryError as e:
             if returnempty:
                 return repo.scm_instance.EMPTY_CHANGESET
             h.flash(_('There are no changesets yet'), category='error')
             raise webob.exc.HTTPNotFound()
         except ChangesetDoesNotExistError as e:
             h.flash(_('Changeset for %s %s not found in %s') %
                               (ref_type, ref_name, repo.repo_name),
                     category='error')
             raise webob.exc.HTTPNotFound()
         except RepositoryError as e:
             log.error(traceback.format_exc())
-            h.flash(safe_str(e), category='error')
+            h.flash(unicode(e), category='error')
             raise webob.exc.HTTPBadRequest()
 @decorator.decorator
 def jsonify(func, *args, **kwargs):
     """Action decorator that formats output for JSON
     Given a function that will return content, this decorator will turn
     the result into JSON, with a content-type of 'application/json' and
     output it.
     """
     response.headers['Content-Type'] = 'application/json; charset=utf-8'
     data = func(*args, **kwargs)
     if isinstance(data, (list, tuple)):
         # A JSON list response is syntactically valid JavaScript and can be
         # loaded and executed as JavaScript by a malicious third-party site
         # using <script>, which can lead to cross-site data leaks.
         # JSON responses should therefore be scalars or objects (i.e. Python
         # dicts), because a JSON object is a syntax error if intepreted as JS.
         msg = "JSON responses with Array envelopes are susceptible to " \
               "cross-site data leak attacks, see " \
               "https://web.archive.org/web/20120519231904/http://wiki.pylonshq.com/display/pylonsfaq/Warnings"
         warnings.warn(msg, Warning, 2)
         log.warning(msg)
     log.debug("Returning JSON wrapped action output")
     return json.dumps(data)
 @decorator.decorator
 def IfSshEnabled(func, *args, **kwargs):
     """Decorator for functions that can only be called if SSH access is enabled.
     If SSH access is disabled in the configuration file, HTTPNotFound is raised.
     """
     if not c.ssh_enabled:
         from kallithea.lib import helpers as h
         h.flash(_("SSH access is disabled."), category='warning')
         raise webob.exc.HTTPNotFound()
     return func(*args, **kwargs)

kallithea/lib/vcs/utils/__init__.py

➞

Show inline comments

 """
 This module provides some useful tools for ``vcs`` like annotate/diff html
 output. It also includes some internal helpers.
 """
 import datetime
 import re
 import time
 def makedate():
     lt = time.localtime()
     if lt[8] == 1 and time.daylight:
         tz = time.altzone
     else:
         tz = time.timezone
     return time.mktime(lt), tz
 def aslist(obj, sep=None, strip=True):
     """
     Returns given string separated by sep as list
     :param obj:
     :param sep:
     :param strip:
     """
     if isinstance(obj, (basestring)):
         lst = obj.split(sep)
         if strip:
             lst = [v.strip() for v in lst]
         return lst
     elif isinstance(obj, (list, tuple)):
         return obj
     elif obj is None:
         return []
     else:
         return [obj]
 def date_fromtimestamp(unixts, tzoffset=0):
     """
     Makes a local datetime object out of unix timestamp
     :param unixts:
     :param tzoffset:
     """
     return datetime.datetime.fromtimestamp(float(unixts))
 def safe_int(val, default=None):
     """
     Returns int() of val if val is not convertible to int use default
     instead
     :param val:
     :param default:
     """
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
     return val
 def safe_unicode(s):
     """
     Safe unicode function. Use a few tricks to turn s into unicode string:
     In case of UnicodeDecodeError with configured default encodings, try to
     detect encoding with chardet library, then fall back to first encoding with
     errors replaced.
     """
     if isinstance(s, unicode):
         return s
     if not isinstance(s, str):  # use __str__ / __unicode__ and don't expect UnicodeDecodeError
         return unicode(s)
     from kallithea.lib.vcs.conf import settings
     for enc in settings.DEFAULT_ENCODINGS:
         try:
             return unicode(s, enc)
         except UnicodeDecodeError:
             pass
     try:
         import chardet
         encoding = chardet.detect(s)['encoding']
         if encoding is not None:
             return s.decode(encoding)
     except (ImportError, UnicodeDecodeError):
         pass
     return unicode(s, settings.DEFAULT_ENCODINGS[0], 'replace')
 def safe_str(s):
     """
     Safe str function. Use a few tricks to turn s into bytes string:
     In case of UnicodeEncodeError with configured default encodings, fall back
     to first configured encoding with errors replaced.
     """
     if isinstance(s, str):
         return s
     if not isinstance(s, unicode):
         return str(s)
     assert isinstance(s, unicode), s  # don't use safe_str to coerce non-strings
     from kallithea.lib.vcs.conf import settings
     for enc in settings.DEFAULT_ENCODINGS:
         try:
             return s.encode(enc)
         except UnicodeEncodeError:
             pass
     return s.encode(settings.DEFAULT_ENCODINGS[0], 'replace')
 # Regex taken from http://www.regular-expressions.info/email.html
 email_re = re.compile(
     r"""[a-z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[a-z0-9!#$%&'*+/=?^_`{|}~-]+)*@"""
     r"""(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]*[a-z0-9])?""",
     re.IGNORECASE)
 def author_email(author):
     """
     Returns email address of given author string.
     If author contains <> brackets, only look inside that.
     If any RFC valid email address is found, return that.
     Else, return empty string.
     """
     if not author:
         return ''
     l = author.find('<') + 1
     if l != 0:
         r = author.find('>', l)
         if r != -1:
             author = author[l:r]
     m = email_re.search(author)
     if m is None:
         return ''
     return safe_str(m.group(0))
 def author_name(author):
     """
     get name of author, or else username.
     It'll try to find an email in the author string and just cut it off
     to get the username
     """
     if not author:
         return ''
     if '@' not in author:
         return author
     return author.replace(author_email(author), '').replace('<', '') \
         .replace('>', '').strip()

0 comments (0 inline, 0 general)