:license: GPLv3, see LICENSE.md for more details.

"""

from pygments import highlight

from pygments.formatters import HtmlFormatter

from kallithea.lib.vcs.exceptions import VCSError

from kallithea.lib.vcs.nodes import FileNode

from kallithea.lib.vcs.utils import safe_str

def annotate_highlight(filenode, annotate_from_changeset_func,

    :param order: ordered sequence of ``ls`` (line numbers column),

      ``annotate`` (annotate column), ``code`` (code column); Default is

      ``['ls', 'annotate', 'code']``

    :param headers: dictionary with headers (keys are whats in ``order``

      parameter)

    """

    options['linenos'] = True

    formatter = AnnotateHtmlFormatter(filenode=filenode,

        annotate_from_changeset_func=annotate_from_changeset_func, order=order,

        headers=headers, **options)

    lexer = get_custom_lexer(filenode.extension) or filenode.lexer

    highlighted = highlight(safe_str(filenode.content), lexer, formatter)

Authentication modules

"""

import importlib

import logging

import traceback

from kallithea.lib.auth import AuthUser, PasswordGenerator

from kallithea.lib.compat import hybrid_property

from kallithea.lib.utils2 import asbool

from kallithea.model.db import Setting, User

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

    def __init__(self, formencode_obj, *args, **kwargs):

        self.formencode_obj = formencode_obj

        self.args = args

        self.kwargs = kwargs

    def __call__(self, *args, **kwargs):

        formencode_obj = self.formencode_obj

        if isfunction(formencode_obj):

            # case we wrap validators into functions

            formencode_obj = self.formencode_obj(*args, **kwargs)

        return formencode_obj(*self.args, **self.kwargs)

        # translator object _() not being registered properly.

        class LazyCaller(object):

            def __init__(self, name):

                self.validator_name = name

            def __call__(self, *args, **kwargs):

                #log.debug('Initializing lazy formencode object: %s', obj)

                return LazyFormencode(obj, *args, **kwargs)

        class ProxyGet(object):

            def __getattribute__(self, name):

                return LazyCaller(name)

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

from kallithea.lib.compat import hybrid_property

log = logging.getLogger(__name__)

            "email": userobj.email,

            "admin": userobj.admin,

            "extern_name": userobj.user_id,

        }

        log.debug('user data: %s', user_data)

        password_match = auth.check_password(password, userobj.password)

        if userobj.is_default_user:

            log.info('user %s authenticated correctly as anonymous user',

                     username)

            return user_data

from time import mktime

import celery.utils.log

from tg import config

import kallithea

from kallithea.lib import celerylib, ext_json

from kallithea.lib.helpers import person

from kallithea.lib.hooks import log_create_repository

from kallithea.lib.rcmail.smtp_mailer import SmtpMailer

from kallithea.lib.utils import action_logger

from kallithea.lib.utils2 import asbool, ascii_bytes

from kallithea.lib.vcs.utils import author_email

__all__ = ['whoosh_index', 'get_commits_stats', 'send_email']

log = celery.utils.log.get_task_logger(__name__)

@celerylib.task

@celerylib.locked_task

@celerylib.dbsession

def whoosh_index(repo_location, full_index):

    celerylib.get_session() # initialize database connection

    index_location = config['index_dir']

    WhooshIndexingDaemon(index_location=index_location,

                         repo_location=repo_location) \

                         .run(full_index=full_index)

    return True

@celerylib.task

@celerylib.dbsession

def create_repo(form_data, cur_user):

    DBS = celerylib.get_session()

    cur_user = User.guess_instance(cur_user)

    owner = cur_user

    repo_name = form_data['repo_name']

    """

    Creates a fork of repository using interval VCS methods

    :param form_data:

    :param cur_user:

    """

    DBS = celerylib.get_session()

    base_path = kallithea.CONFIG['base_path']

    cur_user = User.guess_instance(cur_user)

    repo_name = form_data['repo_name']  # fork in this case

        raise

    return True

def __get_codes_stats(repo_name):

    repo = Repository.get_by_repo_name(repo_name).scm_instance

    tip = repo.get_changeset()

    code_stats = {}

    for _topnode, _dirnodes, filenodes in tip.walk('/'):

        for filenode in filenodes:

            ext = filenode.extension.lower()

                if ext in code_stats:

                    code_stats[ext] += 1

                else:

                    code_stats[ext] = 1

    return code_stats or {}

import alembic.command

import alembic.config

import sqlalchemy

from sqlalchemy.engine import create_engine

from kallithea.model.base import init_model

from kallithea.model.db import Repository, Setting, Ui, User

from kallithea.model.meta import Base, Session

from kallithea.model.permission import PermissionModel

from kallithea.model.user import UserModel

    def _ask_ok(self, msg):

        """Invoke ask_ok unless the force_ask option provides the answer"""

        force_ask = self.cli_args.get('force_ask')

        if force_ask is not None:

            return force_ask

        return ask_ok(msg)

    def init_db(self, SESSION=None):

        if SESSION:

            self.sa = SESSION

        else:

import textwrap

import urllib.parse

from beaker.cache import cache_region

from pygments import highlight as code_highlight

from pygments.formatters.html import HtmlFormatter

from tg.i18n import ugettext as _

from webhelpers2.html import HTML, escape, literal

from webhelpers2.html.tags import NotGiven, Option, Options, _input, _make_safe_id_component, checkbox, end_form

from webhelpers2.html.tags import form as insecure_form

from webhelpers2.html.tags import hidden, link_to, password, radio

from webhelpers2.html.tags import select as webhelpers2_select

from kallithea.config.routing import url

from kallithea.lib.annotate import annotate_highlight

#==============================================================================

# PERMS

#==============================================================================

from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel

from kallithea.lib.markup_renderer import url_re

from kallithea.lib.pygmentsutils import get_custom_lexer

from kallithea.lib.utils2 import MENTIONS_REGEX, AttributeDict

from kallithea.lib.utils2 import age as _age

from kallithea.lib.utils2 import asbool, credentials_filter, safe_bytes, safe_int, safe_str, time_to_datetime

from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError

#==============================================================================

# SCM FILTERS available via h.

#==============================================================================

from kallithea.lib.vcs.utils import author_email, author_name

# mute pyflakes "imported but unused"

assert Option

assert checkbox

assert end_form

def _session_flash_messages(append=None, clear=False):

    """Manage a message queue in tg.session: return the current message queue

    after appending the given message, and possibly clearing the queue."""

    key = 'flash'

    if key in session:

        flash_messages = session[key]

    else:

        if append is None:  # common fast path - also used for clearing empty queue

            return []  # don't bother saving

        flash_messages = []

@cache_region('long_term', 'user_attr_or_none')

def user_attr_or_none(author, show_attr):

    """Try to match email part of VCS committer string with a local user and return show_attr

    - or return None if user not found"""

    email = author_email(author)

    if email:

        user = User.get_by_email(email)

        if user is not None:

            return getattr(user, show_attr)

    return None

    return None

def person(author, show_attr="username"):

    """Find the user identified by 'author', return one of the users attributes,

    default to the username attribute, None if there is no user"""

    # if author is already an instance use it for extraction

    if isinstance(author, User):

        return getattr(author, show_attr)

    value = user_attr_or_none(author, show_attr)

    if value is not None:

    # Still nothing?  Just pass back the author name if any, else the email

    return author_name(author) or email(author)

def person_by_id(id_, show_attr="username"):

    # maybe it's an ID ?

    if str(id_).isdigit() or isinstance(id_, int):

        id_ = int(id_)

        user = User.get(id_)

        if user is not None:

            return getattr(user, show_attr)

    def get_users_group():

        group_name = action_params

        return group_name

    def get_pull_request():

        pull_request_id = action_params

        nice_id = PullRequest.make_nice_id(pull_request_id)

        deleted = user_log.repository is None

        if deleted:

            repo_name = user_log.repository_name

    _def = 'anonymous@kallithea-scm.org'  # default gravatar

    email_address = email_address or _def

    if email_address == _def:

        return default

    parsed_url = urllib.parse.urlparse(url.current(qualified=True))

    return (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL) \

               .replace('{email}', email_address) \

               .replace('{md5email}', hashlib.md5(safe_bytes(email_address).lower()).hexdigest()) \

               .replace('{netloc}', parsed_url.netloc) \

               .replace('{scheme}', parsed_url.scheme) \

    """

    Displays a fancy two colored bar for number of added/deleted

    lines of code on file

    :param stats: two element list of added/deleted lines of code

    """

    a, d = stats['added'], stats['deleted']

    width = 100

    if stats['binary']:

        # binary mode

def urlify_issues(newtext, repo_name):

    """Urlify issue references according to .ini configuration"""

    global _urlify_issues_f

    if _urlify_issues_f is None:

        assert kallithea.CONFIG['sqlalchemy.url'] # make sure config has been loaded

        # Build chain of urlify functions, starting with not doing any transformation

        def tmp_urlify_issues_f(s):

            return s

    if rev and ref_type != 'rev':

        l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))

    return l

def changeset_status(repo, revision):

    return ChangesetStatusModel().get_status(repo, revision)

def changeset_status_lbl(changeset_status):

    return ChangesetStatus.get_status_lbl(changeset_status)

def get_permission_name(key):

    return dict(Permission.PERMS).get(key)

def journal_filter_help():

    return _(textwrap.dedent('''

        Example filter terms:

            ' it was created or renamed from the filesystem'

            ' please run the application again'

            ' in order to rescan repositories') % repo_name, category='error')

def ip_range(ip_addr):

    s, e = UserIpMap._get_ip_range(ip_addr)

    return '%s - %s' % (s, e)

session_csrf_secret_name = "_session_csrf_secret_token"

def session_csrf_secret_token():

    """Return (and create) the current session's CSRF protection token."""

    if not session_csrf_secret_name in session:

        session[session_csrf_secret_name] = str(random.getrandbits(128))

        session.save()

    return session[session_csrf_secret_name]

def form(url, method="post", **attrs):

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

import re

import traceback

import bleach

import markdown as markdown_mod

from kallithea.lib.utils2 import MENTIONS_REGEX, safe_str

log = logging.getLogger(__name__)

    def _flavored_markdown(cls, text):

        """

        Github style flavored markdown

        :param text:

        """

        # Extract pre blocks.

        extractions = {}

        def pre_extraction_callback(matchobj):

            extractions[digest] = matchobj.group(0)

            return "{gfm-extraction-%s}" % digest

        pattern = re.compile(r'<pre>.*?</pre>', re.MULTILINE | re.DOTALL)

        text = re.sub(pattern, pre_extraction_callback, text)

        # Prevent foo_bar_baz from ending up with an italic word in the middle.

                raise

    @classmethod

    def rst(cls, source, safe=True):

        source = safe_str(source)

        try:

            docutils_settings = dict([(alias, None) for alias in

                                cls.RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES])

            docutils_settings.update({'input_encoding': 'unicode',

                                      'report_level': 4})

            parts = publish_parts(source=source,

                                  writer_name="html4css1",

                                  settings_overrides=docutils_settings)

            return parts['html_title'] + parts["fragment"]

        except Exception:

            log.error(traceback.format_exc())

            if safe:

                log.debug('Falling back to render in plain mode')

                return cls.plain(source)

            else:

"""

from collections import defaultdict

from pygments import lexers

def get_extension_descriptions():

    """

    Based on what's inside pygments lexers, return a mapping from lowercase

    extensions to lists of very brief descriptions.

    """

def get_custom_lexer(extension):

    """

    returns a custom lexer if it's defined in the extensions module, or None

    if there's no custom lexer defined

    """

    lexer_name = getattr(kallithea.EXTENSIONS, 'EXTRA_LEXERS', {}).get(extension)

    if lexer_name is None:

        return None

    return lexers.get_lexer_by_name(lexer_name)

import os

import re

import time

import urllib.parse

import urlobject

from tg.i18n import ugettext as _

from tg.i18n import ungettext

from tg.support.converters import asbool, aslist

from webhelpers2.text import collapse, remove_formatting, strip_tags

import kallithea

from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export

from kallithea.lib.vcs.utils.lazy import LazyProperty

try:

    import pwd

    if prevdate > now:

        now, prevdate = prevdate, now

        future = True

    if future:

        prevdate = prevdate.replace(microsecond=0)

    # Get date parts deltas

    for part in order:

        d = relativedelta.relativedelta(now, prevdate)

        deltas[part] = getattr(d, part + 's')

    # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,

    # not 1 hour, -59 minutes and -59 seconds)

    Safe version of get_changeset if this changeset doesn't exists for a

    repo it returns a Dummy one instead

    :param repo:

    :param rev:

    """

    if not isinstance(repo, BaseRepository):

        raise Exception('You must pass an Repository '

                        'object as first argument got %s' % type(repo))

    try:

        cs = repo.get_changeset(rev)

        return self.get(attr, None)

    __setattr__ = dict.__setitem__

    __delattr__ = dict.__delitem__

def obfuscate_url_pw(engine):

    try:

        _url = sa_url.make_url(engine or '')

    except ArgumentError:

        return engine

    if _url.password:

        _url.password = 'XXXXX'

    :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.

"""

import datetime

import itertools

from kallithea.lib.vcs.conf import settings

from kallithea.lib.vcs.exceptions import (ChangesetError, EmptyRepositoryError, NodeAlreadyAddedError, NodeAlreadyChangedError, NodeAlreadyExistsError,

                                          NodeAlreadyRemovedError, NodeDoesNotExistError, NodeNotChangedError, RepositoryError)

from kallithea.lib.vcs.utils import author_email, author_name

from kallithea.lib.vcs.utils.helpers import get_dict_for_attrs

from kallithea.lib.vcs.utils.lazy import LazyProperty

        """

        return self._empty_cs

    @LazyProperty

    def branch(self):

        return get_backend(self.alias).DEFAULT_BRANCH_NAME

    @LazyProperty

    def branches(self):

        return [get_backend(self.alias).DEFAULT_BRANCH_NAME]

    @LazyProperty

    def short_id(self):

        return self.raw_id[:12]

import functools

import mimetypes

import posixpath

import stat

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.lib.vcs.exceptions import NodeError, RemovedFileNodeError

from kallithea.lib.vcs.utils import safe_bytes, safe_str

from kallithea.lib.vcs.utils.lazy import LazyProperty

                encoding = None

            else:

                mtype = 'text/plain'

                encoding = None

                # try with pygments

                try:

                    mt = lexers.get_lexer_for_filename(self.name).mimetypes

                except lexers.ClassNotFound:

                    mt = None

                if mt:

    @LazyProperty

    def lexer(self):

        """

        Returns pygment's lexer class. Would try to guess lexer taking file's

        content, name and mimetype.

        """

        try:

            lexer = lexers.guess_lexer_for_filename(self.name, safe_str(self.content), stripnl=False)

        except lexers.ClassNotFound:

            lexer = lexers.TextLexer(stripnl=False)

        # returns first alias

        return lexer

"""

import datetime

import re

import time

def makedate():

    lt = time.localtime()

    if lt[8] == 1 and time.daylight:

        tz = time.altzone

    else:

    if isinstance(s, str):

        return s

    if not isinstance(s, bytes):  # use __str__ and don't expect UnicodeDecodeError

        return str(s)

    for enc in settings.DEFAULT_ENCODINGS:

        try:

            return str(s, enc)

        except UnicodeDecodeError:

            pass

    try:

        encoding = chardet.detect(s)['encoding']

        if encoding is not None:

            return s.decode(encoding)

        pass

    return str(s, settings.DEFAULT_ENCODINGS[0], 'replace')

def safe_bytes(s):

    """

    if isinstance(s, bytes):

        return s

    assert isinstance(s, str), repr(s)  # bytes cannot coerse with __str__ or handle None or int

    for enc in settings.DEFAULT_ENCODINGS:

        try:

            return s.encode(enc)

        except UnicodeEncodeError:

            pass

"""

Utilities aimed to help achieve mostly basic tasks.

"""

import datetime

import os

import re

import time

import urllib.request

import mercurial.url

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError

from kallithea.lib.vcs.utils import safe_str

from kallithea.lib.vcs.utils.paths import abspath

def get_highlighted_code(name, code, type='terminal'):

    """

    If pygments are available on the system

    then returned output is colored. Otherwise

    unchanged content is returned.

    """

    try:

        lexer = guess_lexer_for_filename(name, code)

        formatter = TerminalFormatter()

        content = highlight(code, lexer, formatter)

    except ClassNotFound:

        logging.debug("Couldn't guess Lexer, will not use pygments.")