Changeset - 1048307eb1f5
[Not reviewed]
default
0 3 0
timeless@gmail.com - 10 years ago 2016-05-03 14:09:01
timeless@gmail.com
spelling: overridden
3 files changed with 3 insertions and 3 deletions:
0 comments (0 inline, 0 general)
docs/setup.rst
Show inline comments
 
.. _setup:
 

	
 
=====
 
Setup
 
=====
 

	
 

	
 
Setting up Kallithea
 
--------------------
 

	
 
First, you will need to create a Kallithea configuration file. Run the
 
following command to do so::
 

	
 
    paster make-config Kallithea my.ini
 

	
 
This will create the file ``my.ini`` in the current directory. This
 
configuration file contains the various settings for Kallithea, e.g.
 
proxy port, email settings, usage of static files, cache, Celery
 
settings, and logging.
 

	
 
Next, you need to create the databases used by Kallithea. It is recommended to
 
use PostgreSQL or SQLite (default). If you choose a database other than the
 
default, ensure you properly adjust the database URL in your ``my.ini``
 
configuration file to use this other database. Kallithea currently supports
 
PostgreSQL, SQLite and MySQL databases. Create the database by running
 
the following command::
 

	
 
    paster setup-db my.ini
 

	
 
This will prompt you for a "root" path. This "root" path is the location where
 
Kallithea will store all of its repositories on the current machine. After
 
entering this "root" path ``setup-db`` will also prompt you for a username
 
and password for the initial admin account which ``setup-db`` sets
 
up for you.
 

	
 
The ``setup-db`` values can also be given on the command line.
 
Example::
 

	
 
    paster setup-db my.ini --user=nn --password=secret --email=nn@example.com --repos=/srv/repos
 

	
 
The ``setup-db`` command will create all needed tables and an
 
admin account. When choosing a root path you can either use a new
 
empty location, or a location which already contains existing
 
repositories. If you choose a location which contains existing
 
repositories Kallithea will add all of the repositories at the chosen
 
location to its database.  (Note: make sure you specify the correct
 
path to the root).
 

	
 
.. note:: the given path for Mercurial_ repositories **must** be write
 
          accessible for the application. It's very important since
 
          the Kallithea web interface will work without write access,
 
          but when trying to do a push it will fail with permission
 
          denied errors unless it has write access.
 

	
 
You are now ready to use Kallithea. To run it simply execute::
 

	
 
    paster serve my.ini
 

	
 
- This command runs the Kallithea server. The web app should be available at
 
  http://127.0.0.1:5000. The IP address and port is configurable via the
 
  configuration file created in the previous step.
 
- Log in to Kallithea using the admin account created when running ``setup-db``.
 
- The default permissions on each repository is read, and the owner is admin.
 
  Remember to update these if needed.
 
- In the admin panel you can toggle LDAP, anonymous, and permissions
 
  settings, as well as edit more advanced options on users and
 
  repositories.
 

	
 

	
 
Using Kallithea with SSH
 
------------------------
 

	
 
Kallithea currently only hosts repositories using http and https. (The addition
 
of ssh hosting is a planned future feature.) However you can easily use ssh in
 
parallel with Kallithea. (Repository access via ssh is a standard "out of
 
the box" feature of Mercurial_ and you can use this to access any of the
 
repositories that Kallithea is hosting. See PublishingRepositories_)
 

	
 
Kallithea repository structures are kept in directories with the same name
 
as the project. When using repository groups, each group is a subdirectory.
 
This allows you to easily use ssh for accessing repositories.
 

	
 
In order to use ssh you need to make sure that your web server and the users'
 
login accounts have the correct permissions set on the appropriate directories.
 

	
 
.. note:: These permissions are independent of any permissions you
 
          have set up using the Kallithea web interface.
 

	
 
If your main directory (the same as set in Kallithea settings) is for
 
example set to ``/srv/repos`` and the repository you are using is
 
named ``kallithea``, then to clone via ssh you should run::
 

	
 
    hg clone ssh://user@kallithea.example.com/srv/repos/kallithea
 

	
 
Using other external tools such as mercurial-server_ or using ssh key-based
 
authentication is fully supported.
 

	
 
.. note:: In an advanced setup, in order for your ssh access to use
 
          the same permissions as set up via the Kallithea web
 
          interface, you can create an authentication hook to connect
 
          to the Kallithea db and run check functions for permissions
 
          against that.
 

	
 

	
 
Setting up Whoosh full text search
 
----------------------------------
 

	
 
Kallithea provides full text search of repositories using `Whoosh`__.
 

	
 
.. __: https://pythonhosted.org/Whoosh/
 

	
 
For an incremental index build, run::
 

	
 
    paster make-index my.ini
 

	
 
For a full index rebuild, run::
 

	
 
    paster make-index my.ini -f
 

	
 
The ``--repo-location`` option allows the location of the repositories to be overriden;
 
The ``--repo-location`` option allows the location of the repositories to be overridden;
 
usually, the location is retrieved from the Kallithea database.
 

	
 
The ``--index-only`` option can be used to limit the indexed repositories to a comma-separated list::
 

	
 
    paster make-index my.ini --index-only=vcs,kallithea
 

	
 
To keep your index up-to-date it is necessary to do periodic index builds;
 
for this, it is recommended to use a crontab entry. Example::
 

	
 
    0  3  *  *  *  /path/to/virtualenv/bin/paster make-index /path/to/kallithea/my.ini
 

	
 
When using incremental mode (the default), Whoosh will check the last
 
modification date of each file and add it to be reindexed if a newer file is
 
available. The indexing daemon checks for any removed files and removes them
 
from index.
 

	
 
If you want to rebuild the index from scratch, you can use the ``-f`` flag as above,
 
or in the admin panel you can check the "build from scratch" checkbox.
 

	
 
.. _ldap-setup:
 

	
 

	
 
Setting up LDAP support
 
-----------------------
 

	
 
Kallithea supports LDAP authentication. In order
 
to use LDAP, you have to install the python-ldap_ package. This package is
 
available via PyPI, so you can install it by running::
 

	
 
    pip install python-ldap
 

	
 
.. note:: ``python-ldap`` requires some libraries to be installed on
 
          your system, so before installing it check that you have at
 
          least the ``openldap`` and ``sasl`` libraries.
 

	
 
Choose *Admin > Authentication*, click the ``kallithea.lib.auth_modules.auth_ldap`` button
 
and then *Save*, to enable the LDAP plugin and configure its settings.
 

	
 
Here's a typical LDAP setup::
 

	
 
 Connection settings
 
 Enable LDAP          = checked
 
 Host                 = host.example.com
 
 Port                 = 389
 
 Account              = <account>
 
 Password             = <password>
 
 Connection Security  = LDAPS connection
 
 Certificate Checks   = DEMAND
 

	
 
 Search settings
 
 Base DN              = CN=users,DC=host,DC=example,DC=org
 
 LDAP Filter          = (&(objectClass=user)(!(objectClass=computer)))
 
 LDAP Search Scope    = SUBTREE
 

	
 
 Attribute mappings
 
 Login Attribute      = uid
 
 First Name Attribute = firstName
 
 Last Name Attribute  = lastName
 
 Email Attribute      = mail
 

	
 
If your user groups are placed in an Organisation Unit (OU) structure, the Search Settings configuration differs::
 

	
 
 Search settings
 
 Base DN              = DC=host,DC=example,DC=org
 
 LDAP Filter          = (&(memberOf=CN=your user group,OU=subunit,OU=unit,DC=host,DC=example,DC=org)(objectClass=user))
 
 LDAP Search Scope    = SUBTREE
 

	
 
.. _enable_ldap:
 

	
 
Enable LDAP : required
 
    Whether to use LDAP for authenticating users.
 

	
 
.. _ldap_host:
 

	
 
Host : required
 
    LDAP server hostname or IP address. Can be also a comma separated
 
    list of servers to support LDAP fail-over.
 

	
 
.. _Port:
 

	
 
Port : required
 
    389 for un-encrypted LDAP, 636 for SSL-encrypted LDAP.
 

	
 
.. _ldap_account:
 

	
 
Account : optional
 
    Only required if the LDAP server does not allow anonymous browsing of
 
    records.  This should be a special account for record browsing.  This
 
    will require `LDAP Password`_ below.
 

	
 
.. _LDAP Password:
 

	
 
Password : optional
 
    Only required if the LDAP server does not allow anonymous browsing of
 
    records.
 

	
 
.. _Enable LDAPS:
 

	
 
Connection Security : required
 
    Defines the connection to LDAP server
 

	
 
    No encryption
 
        Plain non encrypted connection
 

	
 
    LDAPS connection
 
        Enable LDAPS connections. It will likely require `Port`_ to be set to
 
        a different value (standard LDAPS port is 636). When LDAPS is enabled
 
        then `Certificate Checks`_ is required.
 

	
 
    START_TLS on LDAP connection
 
        START TLS connection
 

	
 
.. _Certificate Checks:
 

	
 
Certificate Checks : optional
 
    How SSL certificates verification is handled -- this is only useful when
 
    `Enable LDAPS`_ is enabled.  Only DEMAND or HARD offer full SSL security
 
    while the other options are susceptible to man-in-the-middle attacks.  SSL
 
    certificates can be installed to /etc/openldap/cacerts so that the
 
    DEMAND or HARD options can be used with self-signed certificates or
 
    certificates that do not have traceable certificates of authority.
 

	
 
    NEVER
 
        A serve certificate will never be requested or checked.
 

	
 
    ALLOW
 
        A server certificate is requested.  Failure to provide a
 
        certificate or providing a bad certificate will not terminate the
 
        session.
 

	
 
    TRY
 
        A server certificate is requested.  Failure to provide a
 
        certificate does not halt the session; providing a bad certificate
 
        halts the session.
 

	
 
    DEMAND
 
        A server certificate is requested and must be provided and
 
        authenticated for the session to proceed.
 

	
 
    HARD
 
        The same as DEMAND.
 

	
 
.. _Base DN:
 

	
 
Base DN : required
 
    The Distinguished Name (DN) where searches for users will be performed.
 
    Searches can be controlled by `LDAP Filter`_ and `LDAP Search Scope`_.
 

	
 
.. _LDAP Filter:
 

	
 
LDAP Filter : optional
 
    A LDAP filter defined by RFC 2254.  This is more useful when `LDAP
 
    Search Scope`_ is set to SUBTREE.  The filter is useful for limiting
 
    which LDAP objects are identified as representing Users for
 
    authentication.  The filter is augmented by `Login Attribute`_ below.
 
    This can commonly be left blank.
 

	
 
.. _LDAP Search Scope:
 

	
 
LDAP Search Scope : required
 
    This limits how far LDAP will search for a matching object.
 

	
 
    BASE
 
        Only allows searching of `Base DN`_ and is usually not what you
 
        want.
 

	
 
    ONELEVEL
 
        Searches all entries under `Base DN`_, but not Base DN itself.
 

	
 
    SUBTREE
 
        Searches all entries below `Base DN`_, but not Base DN itself.
 
        When using SUBTREE `LDAP Filter`_ is useful to limit object
 
        location.
 

	
 
.. _Login Attribute:
 

	
 
Login Attribute : required
 
    The LDAP record attribute that will be matched as the USERNAME or
 
    ACCOUNT used to connect to Kallithea.  This will be added to `LDAP
 
    Filter`_ for locating the User object.  If `LDAP Filter`_ is specified as
 
    "LDAPFILTER", `Login Attribute`_ is specified as "uid" and the user has
 
    connected as "jsmith" then the `LDAP Filter`_ will be augmented as below
 
    ::
 

	
 
        (&(LDAPFILTER)(uid=jsmith))
 

	
 
.. _ldap_attr_firstname:
 

	
 
First Name Attribute : required
 
    The LDAP record attribute which represents the user's first name.
 

	
 
.. _ldap_attr_lastname:
 

	
 
Last Name Attribute : required
 
    The LDAP record attribute which represents the user's last name.
 

	
 
.. _ldap_attr_email:
 

	
 
Email Attribute : required
 
    The LDAP record attribute which represents the user's email address.
 

	
 
If all data are entered correctly, and python-ldap_ is properly installed
 
users should be granted access to Kallithea with LDAP accounts.  At this
 
time user information is copied from LDAP into the Kallithea user database.
 
This means that updates of an LDAP user object may not be reflected as a
 
user update in Kallithea.
 

	
 
If You have problems with LDAP access and believe You entered correct
 
information check out the Kallithea logs, any error messages sent from LDAP
 
will be saved there.
 

	
 
Active Directory
 
^^^^^^^^^^^^^^^^
 

	
 
Kallithea can use Microsoft Active Directory for user authentication.  This
 
is done through an LDAP or LDAPS connection to Active Directory.  The
 
following LDAP configuration settings are typical for using Active
 
Directory ::
 

	
 
 Base DN              = OU=SBSUsers,OU=Users,OU=MyBusiness,DC=v3sys,DC=local
 
 Login Attribute      = sAMAccountName
 
 First Name Attribute = givenName
 
 Last Name Attribute  = sn
 
 Email Attribute     = mail
 

	
 
All other LDAP settings will likely be site-specific and should be
 
appropriately configured.
 

	
 

	
 
Authentication by container or reverse-proxy
 
--------------------------------------------
 

	
 
Kallithea supports delegating the authentication
 
of users to its WSGI container, or to a reverse-proxy server through which all
 
clients access the application.
 

	
 
When these authentication methods are enabled in Kallithea, it uses the
 
username that the container/proxy (Apache or Nginx, etc.) provides and doesn't
 
perform the authentication itself. The authorization, however, is still done by
 
Kallithea according to its settings.
 

	
 
When a user logs in for the first time using these authentication methods,
 
a matching user account is created in Kallithea with default permissions. An
 
administrator can then modify it using Kallithea's admin interface.
 

	
 
It's also possible for an administrator to create accounts and configure their
 
permissions before the user logs in for the first time, using the :ref:`create-user` API.
 

	
 
Container-based authentication
 
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

	
 
In a container-based authentication setup, Kallithea reads the user name from
 
the ``REMOTE_USER`` server variable provided by the WSGI container.
 

	
 
After setting up your container (see `Apache with mod_wsgi`_), you'll need
 
to configure it to require authentication on the location configured for
 
Kallithea.
 

	
 
Proxy pass-through authentication
 
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

	
 
In a proxy pass-through authentication setup, Kallithea reads the user name
 
from the ``X-Forwarded-User`` request header, which should be configured to be
 
sent by the reverse-proxy server.
 

	
 
After setting up your proxy solution (see `Apache virtual host reverse proxy example`_,
 
`Apache as subdirectory`_ or `Nginx virtual host example`_), you'll need to
 
configure the authentication and add the username in a request header named
 
``X-Forwarded-User``.
 

	
 
For example, the following config section for Apache sets a subdirectory in a
 
reverse-proxy setup with basic auth:
 

	
 
.. code-block:: apache
 

	
 
    <Location /someprefix>
 
      ProxyPass http://127.0.0.1:5000/someprefix
 
      ProxyPassReverse http://127.0.0.1:5000/someprefix
 
      SetEnvIf X-Url-Scheme https HTTPS=1
 

	
 
      AuthType Basic
 
      AuthName "Kallithea authentication"
 
      AuthUserFile /srv/kallithea/.htpasswd
 
      Require valid-user
 

	
 
      RequestHeader unset X-Forwarded-User
 

	
 
      RewriteEngine On
 
      RewriteCond %{LA-U:REMOTE_USER} (.+)
 
      RewriteRule .* - [E=RU:%1]
 
      RequestHeader set X-Forwarded-User %{RU}e
 
    </Location>
 

	
 
Setting metadata in container/reverse-proxy
 
"""""""""""""""""""""""""""""""""""""""""""
 
When a new user account is created on the first login, Kallithea has no information about
 
the user's email and full name. So you can set some additional request headers like in the
 
example below. In this example the user is authenticated via Kerberos and an Apache
 
mod_python fixup handler is used to get the user information from a LDAP server. But you
 
could set the request headers however you want.
 

	
 
.. code-block:: apache
 

	
 
    <Location /someprefix>
 
      ProxyPass http://127.0.0.1:5000/someprefix
 
      ProxyPassReverse http://127.0.0.1:5000/someprefix
 
      SetEnvIf X-Url-Scheme https HTTPS=1
 

	
 
      AuthName "Kerberos Login"
 
      AuthType Kerberos
 
      Krb5Keytab /etc/apache2/http.keytab
 
      KrbMethodK5Passwd off
 
      KrbVerifyKDC on
 
      Require valid-user
 

	
 
      PythonFixupHandler ldapmetadata
 

	
 
      RequestHeader set X_REMOTE_USER %{X_REMOTE_USER}e
 
      RequestHeader set X_REMOTE_EMAIL %{X_REMOTE_EMAIL}e
 
      RequestHeader set X_REMOTE_FIRSTNAME %{X_REMOTE_FIRSTNAME}e
 
      RequestHeader set X_REMOTE_LASTNAME %{X_REMOTE_LASTNAME}e
 
    </Location>
 

	
 
.. code-block:: python
 

	
 
    from mod_python import apache
 
    import ldap
 

	
 
    LDAP_SERVER = "ldap://server.mydomain.com:389"
 
    LDAP_USER = ""
 
    LDAP_PASS = ""
 
    LDAP_ROOT = "dc=mydomain,dc=com"
 
    LDAP_FILTER = "sAMAccountName=%s"
 
    LDAP_ATTR_LIST = ['sAMAccountName','givenname','sn','mail']
 

	
 
    def fixuphandler(req):
 
        if req.user is None:
 
            # no user to search for
 
            return apache.OK
 
        else:
 
            try:
 
                if('\\' in req.user):
 
                    username = req.user.split('\\')[1]
 
                elif('@' in req.user):
 
                    username = req.user.split('@')[0]
 
                else:
 
                    username = req.user
 
                l = ldap.initialize(LDAP_SERVER)
 
                l.simple_bind_s(LDAP_USER, LDAP_PASS)
 
                r = l.search_s(LDAP_ROOT, ldap.SCOPE_SUBTREE, LDAP_FILTER % username, attrlist=LDAP_ATTR_LIST)
 

	
 
                req.subprocess_env['X_REMOTE_USER'] = username
 
                req.subprocess_env['X_REMOTE_EMAIL'] = r[0][1]['mail'][0].lower()
 
                req.subprocess_env['X_REMOTE_FIRSTNAME'] = "%s" % r[0][1]['givenname'][0]
 
                req.subprocess_env['X_REMOTE_LASTNAME'] = "%s" % r[0][1]['sn'][0]
 
            except Exception, e:
 
                apache.log_error("error getting data from ldap %s" % str(e), apache.APLOG_ERR)
 

	
 
            return apache.OK
 

	
 
.. note::
 
   If you enable proxy pass-through authentication, make sure your server is
 
   only accessible through the proxy. Otherwise, any client would be able to
 
   forge the authentication header and could effectively become authenticated
 
   using any account of their liking.
 

	
 

	
 
Integration with issue trackers
 
-------------------------------
 

	
 
Kallithea provides a simple integration with issue trackers. It's possible
 
to define a regular expression that will match an issue ID in commit messages,
 
and have that replaced with a URL to the issue. To enable this simply
 
uncomment the following variables in the ini file::
 

	
 
    issue_pat = (?:^#|\s#)(\w+)
 
    issue_server_link = https://issues.example.com/{repo}/issue/{id}
 
    issue_prefix = #
 

	
 
``issue_pat`` is the regular expression describing which strings in
 
commit messages will be treated as issue references. A match group in
 
parentheses should be used to specify the actual issue id.
 

	
 
The default expression matches issues in the format ``#<number>``, e.g., ``#300``.
 

	
 
Matched issue references are replaced with the link specified in
 
``issue_server_link``. ``{id}`` is replaced with the issue ID, and
 
``{repo}`` with the repository name.  Since the # is stripped away,
 
``issue_prefix`` is prepended to the link text.  ``issue_prefix`` doesn't
 
necessarily need to be ``#``: if you set issue prefix to ``ISSUE-`` this will
 
generate a URL in the format:
 

	
 
.. code-block:: html
 

	
 
  <a href="https://issues.example.com/example_repo/issue/300">ISSUE-300</a>
 

	
 
If needed, more than one pattern can be specified by appending a unique suffix to
 
the variables. For example::
 

	
 
    issue_pat_wiki = (?:wiki-)(.+)
 
    issue_server_link_wiki = https://wiki.example.com/{id}
 
    issue_prefix_wiki = WIKI-
 

	
 
With these settings, wiki pages can be referenced as wiki-some-id, and every
 
such reference will be transformed into:
 

	
 
.. code-block:: html
 

	
 
  <a href="https://wiki.example.com/some-id">WIKI-some-id</a>
 

	
 

	
 
Hook management
 
---------------
 

	
 
Hooks can be managed in similar way to that used in ``.hgrc`` files.
 
To manage hooks, choose *Admin > Settings > Hooks*.
 

	
 
The built-in hooks cannot be modified, though they can be enabled or disabled in the *VCS* section.
 

	
 
To add another custom hook simply fill in the first textbox with
 
``<name>.<hook_type>`` and the second with the hook path. Example hooks
 
can be found in ``kallithea.lib.hooks``.
 

	
 

	
 
Changing default encoding
 
-------------------------
 

	
 
By default, Kallithea uses UTF-8 encoding.
 
This is configurable as ``default_encoding`` in the .ini file.
 
This affects many parts in Kallithea including user names, filenames, and
 
encoding of commit messages. In addition Kallithea can detect if the ``chardet``
 
library is installed. If ``chardet`` is detected Kallithea will fallback to it
 
when there are encode/decode errors.
 

	
 

	
 
Celery configuration
 
--------------------
 

	
 
Kallithea can use the distributed task queue system Celery_ to run tasks like
 
cloning repositories or sending emails.
 

	
 
Kallithea will in most setups work perfectly fine out of the box (without
 
Celery), executing all tasks in the web server process. Some tasks can however
 
take some time to run and it can be better to run such tasks asynchronously in
 
a separate process so the web server can focus on serving web requests.
 

	
 
For installation and configuration of Celery, see the `Celery documentation`_.
 
Note that Celery requires a message broker service like RabbitMQ_ (recommended)
 
or Redis_.
 

	
 
The use of Celery is configured in the Kallithea ini configuration file.
 
To enable it, simply set::
 

	
 
  use_celery = true
 

	
 
and add or change the ``celery.*`` and ``broker.*`` configuration variables.
 

	
 
Remember that the ini files use the format with '.' and not with '_' like
 
Celery. So for example setting `BROKER_HOST` in Celery means setting
 
`broker.host` in the configuration file.
 

	
 
To start the Celery process, run::
 

	
 
 paster celeryd <configfile.ini>
 

	
 
.. note::
 
   Make sure you run this command from the same virtualenv, and with the same
 
   user that Kallithea runs.
 

	
 

	
 
HTTPS support
 
-------------
 

	
 
Kallithea will by default generate URLs based on the WSGI environment.
 

	
 
Alternatively, you can use some special configuration settings to control
 
directly which scheme/protocol Kallithea will use when generating URLs:
 

	
 
- With ``https_fixup = true``, the scheme will be taken from the
 
  ``X-Url-Scheme``, ``X-Forwarded-Scheme`` or ``X-Forwarded-Proto`` HTTP header
 
  (default ``http``).
 
- With ``force_https = true`` the default will be ``https``.
 
- With ``use_htsts = true``, Kallithea will set ``Strict-Transport-Security`` when using https.
 

	
 

	
 
Nginx virtual host example
 
--------------------------
 

	
 
Sample config for Nginx using proxy:
 

	
 
.. code-block:: nginx
 

	
 
    upstream kallithea {
 
        server 127.0.0.1:5000;
 
        # add more instances for load balancing
 
        #server 127.0.0.1:5001;
 
        #server 127.0.0.1:5002;
 
    }
 

	
 
    ## gist alias
 
    server {
 
       listen          443;
 
       server_name     gist.example.com;
 
       access_log      /var/log/nginx/gist.access.log;
 
       error_log       /var/log/nginx/gist.error.log;
 

	
 
       ssl on;
 
       ssl_certificate     gist.your.kallithea.server.crt;
 
       ssl_certificate_key gist.your.kallithea.server.key;
 

	
 
       ssl_session_timeout 5m;
 

	
 
       ssl_protocols SSLv3 TLSv1;
 
       ssl_ciphers DHE-RSA-AES256-SHA:DHE-RSA-AES128-SHA:EDH-RSA-DES-CBC3-SHA:AES256-SHA:DES-CBC3-SHA:AES128-SHA:RC4-SHA:RC4-MD5;
 
       ssl_prefer_server_ciphers on;
 

	
 
       rewrite ^/(.+)$ https://kallithea.example.com/_admin/gists/$1;
 
       rewrite (.*)    https://kallithea.example.com/_admin/gists;
 
    }
 

	
 
    server {
 
       listen          443;
 
       server_name     kallithea.example.com
 
       access_log      /var/log/nginx/kallithea.access.log;
 
       error_log       /var/log/nginx/kallithea.error.log;
 

	
 
       ssl on;
 
       ssl_certificate     your.kallithea.server.crt;
 
       ssl_certificate_key your.kallithea.server.key;
 

	
 
       ssl_session_timeout 5m;
 

	
 
       ssl_protocols SSLv3 TLSv1;
 
       ssl_ciphers DHE-RSA-AES256-SHA:DHE-RSA-AES128-SHA:EDH-RSA-DES-CBC3-SHA:AES256-SHA:DES-CBC3-SHA:AES128-SHA:RC4-SHA:RC4-MD5;
 
       ssl_prefer_server_ciphers on;
 

	
 
       ## uncomment root directive if you want to serve static files by nginx
 
       ## requires static_files = false in .ini file
 
       #root /path/to/installation/kallithea/public;
 
       include         /etc/nginx/proxy.conf;
 
       location / {
 
            try_files $uri @kallithea;
 
       }
 

	
 
       location @kallithea {
 
            proxy_pass      http://127.0.0.1:5000;
 
       }
 

	
 
    }
 

	
 
Here's the proxy.conf. It's tuned so it will not timeout on long
 
pushes or large pushes::
 

	
 
    proxy_redirect              off;
 
    proxy_set_header            Host $host;
 
    ## needed for container auth
 
    #proxy_set_header            REMOTE_USER $remote_user;
 
    #proxy_set_header            X-Forwarded-User $remote_user;
 
    proxy_set_header            X-Url-Scheme $scheme;
 
    proxy_set_header            X-Host $http_host;
 
    proxy_set_header            X-Real-IP $remote_addr;
 
    proxy_set_header            X-Forwarded-For $proxy_add_x_forwarded_for;
 
    proxy_set_header            Proxy-host $proxy_host;
 
    proxy_buffering             off;
 
    proxy_connect_timeout       7200;
 
    proxy_send_timeout          7200;
 
    proxy_read_timeout          7200;
 
    proxy_buffers               8 32k;
 
    client_max_body_size        1024m;
 
    client_body_buffer_size     128k;
 
    large_client_header_buffers 8 64k;
 

	
 

	
 
Apache virtual host reverse proxy example
 
-----------------------------------------
 

	
 
Here is a sample configuration file for Apache using proxy:
 

	
 
.. code-block:: apache
 

	
 
    <VirtualHost *:80>
 
            ServerName kallithea.example.com
 

	
 
            <Proxy *>
 
              # For Apache 2.4 and later:
 
              Require all granted
 

	
 
              # For Apache 2.2 and earlier, instead use:
 
              # Order allow,deny
 
              # Allow from all
 
            </Proxy>
 

	
 
            #important !
 
            #Directive to properly generate url (clone url) for pylons
 
            ProxyPreserveHost On
 

	
 
            #kallithea instance
 
            ProxyPass / http://127.0.0.1:5000/
 
            ProxyPassReverse / http://127.0.0.1:5000/
 

	
 
            #to enable https use line below
 
            #SetEnvIf X-Url-Scheme https HTTPS=1
 
    </VirtualHost>
 

	
 
Additional tutorial
 
http://pylonsbook.com/en/1.1/deployment.html#using-apache-to-proxy-requests-to-pylons
 

	
 

	
 
Apache as subdirectory
 
----------------------
 

	
 
Apache subdirectory part:
 

	
 
.. code-block:: apache
 

	
 
    <Location /<someprefix> >
 
      ProxyPass http://127.0.0.1:5000/<someprefix>
 
      ProxyPassReverse http://127.0.0.1:5000/<someprefix>
 
      SetEnvIf X-Url-Scheme https HTTPS=1
 
    </Location>
 

	
 
Besides the regular apache setup you will need to add the following line
 
into ``[app:main]`` section of your .ini file::
 

	
 
    filter-with = proxy-prefix
 

	
 
Add the following at the end of the .ini file::
 

	
 
    [filter:proxy-prefix]
 
    use = egg:PasteDeploy#prefix
 
    prefix = /<someprefix>
 

	
 
then change ``<someprefix>`` into your chosen prefix
 

	
 

	
 
Apache with mod_wsgi
 
--------------------
 

	
 
Alternatively, Kallithea can be set up with Apache under mod_wsgi. For
 
that, you'll need to:
 

	
 
- Install mod_wsgi. If using a Debian-based distro, you can install
 
  the package libapache2-mod-wsgi::
 

	
 
    aptitude install libapache2-mod-wsgi
 

	
 
- Enable mod_wsgi::
 

	
 
    a2enmod wsgi
 

	
 
- Add global Apache configuration to tell mod_wsgi that Python only will be
 
  used in the WSGI processes and shouldn't be initialized in the Apache
 
  processes::
 

	
 
    WSGIRestrictEmbedded On
 

	
 
- Create a wsgi dispatch script, like the one below. Make sure you
 
  check that the paths correctly point to where you installed Kallithea
 
  and its Python Virtual Environment.
 
- Enable the ``WSGIScriptAlias`` directive for the WSGI dispatch script,
 
  as in the following example. Once again, check the paths are
 
  correctly specified.
 

	
 
Here is a sample excerpt from an Apache Virtual Host configuration file:
 

	
 
.. code-block:: apache
 

	
 
    WSGIDaemonProcess kallithea \
 
        threads=4 \
 
        python-home=/srv/kallithea/venv
 
    WSGIProcessGroup kallithea
 
    WSGIScriptAlias / /srv/kallithea/dispatch.wsgi
 
    WSGIPassAuthorization On
 

	
 
Or if using a dispatcher WSGI script with proper virtualenv activation:
 

	
 
.. code-block:: apache
 

	
 
    WSGIDaemonProcess kallithea threads=4
 
    WSGIProcessGroup kallithea
 
    WSGIScriptAlias / /srv/kallithea/dispatch.wsgi
 
    WSGIPassAuthorization On
 

	
 
Apache will by default run as a special Apache user, on Linux systems
 
usually ``www-data`` or ``apache``. If you need to have the repositories
 
directory owned by a different user, use the user and group options to
 
WSGIDaemonProcess to set the name of the user and group.
 

	
 
Example WSGI dispatch script:
 

	
 
.. code-block:: python
 

	
 
    import os
 
    os.environ["HGENCODING"] = "UTF-8"
 
    os.environ['PYTHON_EGG_CACHE'] = '/srv/kallithea/.egg-cache'
 

	
 
    # sometimes it's needed to set the current dir
 
    os.chdir('/srv/kallithea/')
 

	
 
    import site
 
    site.addsitedir("/srv/kallithea/venv/lib/python2.7/site-packages")
 

	
 
    ini = '/srv/kallithea/my.ini'
 
    from paste.script.util.logging_config import fileConfig
 
    fileConfig(ini)
 
    from paste.deploy import loadapp
 
    application = loadapp('config:' + ini)
 

	
 
Or using proper virtualenv activation:
 

	
 
.. code-block:: python
 

	
 
    activate_this = '/srv/kallithea/venv/bin/activate_this.py'
 
    execfile(activate_this, dict(__file__=activate_this))
 

	
 
    import os
 
    os.environ['HOME'] = '/srv/kallithea'
 

	
 
    ini = '/srv/kallithea/kallithea.ini'
 
    from paste.script.util.logging_config import fileConfig
 
    fileConfig(ini)
 
    from paste.deploy import loadapp
 
    application = loadapp('config:' + ini)
 

	
 

	
 
Other configuration files
 
-------------------------
 

	
 
A number of `example init.d scripts`__ can be found in
 
the ``init.d`` directory of the Kallithea source.
 

	
 
.. __: https://kallithea-scm.org/repos/kallithea/files/tip/init.d/ .
 

	
 

	
 
.. _virtualenv: http://pypi.python.org/pypi/virtualenv
 
.. _python: http://www.python.org/
 
.. _Mercurial: http://mercurial.selenic.com/
 
.. _Celery: http://celeryproject.org/
 
.. _Celery documentation: http://docs.celeryproject.org/en/latest/getting-started/index.html
 
.. _RabbitMQ: http://www.rabbitmq.com/
 
.. _Redis: http://redis.io/
 
.. _python-ldap: http://www.python-ldap.org/
 
.. _mercurial-server: http://www.lshift.net/mercurial-server.html
 
.. _PublishingRepositories: http://mercurial.selenic.com/wiki/PublishingRepositories
kallithea/lib/auth.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.lib.auth
 
~~~~~~~~~~~~~~~~~~
 

	
 
authentication and permission libraries
 

	
 
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 4, 2010
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 
import time
 
import os
 
import logging
 
import traceback
 
import hashlib
 
import itertools
 
import collections
 

	
 
from decorator import decorator
 

	
 
from pylons import url, request, session
 
from pylons.i18n.translation import _
 
from webhelpers.pylonslib import secure_form
 
from sqlalchemy import or_
 
from sqlalchemy.orm.exc import ObjectDeletedError
 
from sqlalchemy.orm import joinedload
 
from webob.exc import HTTPFound, HTTPBadRequest, HTTPForbidden, HTTPMethodNotAllowed
 

	
 
from kallithea import __platform__, is_windows, is_unix
 
from kallithea.lib.vcs.utils.lazy import LazyProperty
 
from kallithea.model import meta
 
from kallithea.model.meta import Session
 
from kallithea.model.user import UserModel
 
from kallithea.model.db import User, Repository, Permission, \
 
    UserToPerm, UserGroupRepoToPerm, UserGroupToPerm, UserGroupMember, \
 
    RepoGroup, UserGroupRepoGroupToPerm, UserIpMap, UserGroupUserGroupToPerm, \
 
    UserGroup, UserApiKeys
 

	
 
from kallithea.lib.utils2 import safe_unicode, aslist
 
from kallithea.lib.utils import get_repo_slug, get_repo_group_slug, \
 
    get_user_group_slug, conditional_cache
 
from kallithea.lib.caching_query import FromCache
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class PasswordGenerator(object):
 
    """
 
    This is a simple class for generating password from different sets of
 
    characters
 
    usage::
 

	
 
        passwd_gen = PasswordGenerator()
 
        #print 8-letter password containing only big and small letters
 
            of alphabet
 
        passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
 
    """
 
    ALPHABETS_NUM = r'''1234567890'''
 
    ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
 
    ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
 
    ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
 
    ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
 
        + ALPHABETS_NUM + ALPHABETS_SPECIAL
 
    ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
 
    ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
 
    ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
 
    ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
 

	
 
    def gen_password(self, length, alphabet=ALPHABETS_FULL):
 
        assert len(alphabet) <= 256, alphabet
 
        l = []
 
        while len(l) < length:
 
            i = ord(os.urandom(1))
 
            if i < len(alphabet):
 
                l.append(alphabet[i])
 
        return ''.join(l)
 

	
 

	
 
class KallitheaCrypto(object):
 

	
 
    @classmethod
 
    def hash_string(cls, str_):
 
        """
 
        Cryptographic function used for password hashing based on pybcrypt
 
        or Python's own OpenSSL wrapper on windows
 

	
 
        :param password: password to hash
 
        """
 
        if is_windows:
 
            return hashlib.sha256(str_).hexdigest()
 
        elif is_unix:
 
            import bcrypt
 
            return bcrypt.hashpw(str_, bcrypt.gensalt(10))
 
        else:
 
            raise Exception('Unknown or unsupported platform %s' \
 
                            % __platform__)
 

	
 
    @classmethod
 
    def hash_check(cls, password, hashed):
 
        """
 
        Checks matching password with it's hashed value, runs different
 
        implementation based on platform it runs on
 

	
 
        :param password: password
 
        :param hashed: password in hashed form
 
        """
 

	
 
        if is_windows:
 
            return hashlib.sha256(password).hexdigest() == hashed
 
        elif is_unix:
 
            import bcrypt
 
            return bcrypt.hashpw(password, hashed) == hashed
 
        else:
 
            raise Exception('Unknown or unsupported platform %s' \
 
                            % __platform__)
 

	
 

	
 
def get_crypt_password(password):
 
    return KallitheaCrypto.hash_string(password)
 

	
 

	
 
def check_password(password, hashed):
 
    return KallitheaCrypto.hash_check(password, hashed)
 

	
 

	
 

	
 
def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,
 
                       explicit, algo):
 
    RK = 'repositories'
 
    GK = 'repositories_groups'
 
    UK = 'user_groups'
 
    GLOBAL = 'global'
 
    PERM_WEIGHTS = Permission.PERM_WEIGHTS
 
    permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
 

	
 
    def _choose_perm(new_perm, cur_perm):
 
        new_perm_val = PERM_WEIGHTS[new_perm]
 
        cur_perm_val = PERM_WEIGHTS[cur_perm]
 
        if algo == 'higherwin':
 
            if new_perm_val > cur_perm_val:
 
                return new_perm
 
            return cur_perm
 
        elif algo == 'lowerwin':
 
            if new_perm_val < cur_perm_val:
 
                return new_perm
 
            return cur_perm
 

	
 
    #======================================================================
 
    # fetch default permissions
 
    #======================================================================
 
    default_user = User.get_by_username('default', cache=True)
 
    default_user_id = default_user.user_id
 

	
 
    default_repo_perms = Permission.get_default_perms(default_user_id)
 
    default_repo_groups_perms = Permission.get_default_group_perms(default_user_id)
 
    default_user_group_perms = Permission.get_default_user_group_perms(default_user_id)
 

	
 
    if user_is_admin:
 
        #==================================================================
 
        # admin users have all rights;
 
        # based on default permissions, just set everything to admin
 
        #==================================================================
 
        permissions[GLOBAL].add('hg.admin')
 
        permissions[GLOBAL].add('hg.create.write_on_repogroup.true')
 

	
 
        # repositories
 
        for perm in default_repo_perms:
 
            r_k = perm.UserRepoToPerm.repository.repo_name
 
            p = 'repository.admin'
 
            permissions[RK][r_k] = p
 

	
 
        # repository groups
 
        for perm in default_repo_groups_perms:
 
            rg_k = perm.UserRepoGroupToPerm.group.group_name
 
            p = 'group.admin'
 
            permissions[GK][rg_k] = p
 

	
 
        # user groups
 
        for perm in default_user_group_perms:
 
            u_k = perm.UserUserGroupToPerm.user_group.users_group_name
 
            p = 'usergroup.admin'
 
            permissions[UK][u_k] = p
 
        return permissions
 

	
 
    #==================================================================
 
    # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
 
    #==================================================================
 

	
 
    # default global permissions taken from the default user
 
    default_global_perms = UserToPerm.query() \
 
        .filter(UserToPerm.user_id == default_user_id) \
 
        .options(joinedload(UserToPerm.permission))
 

	
 
    for perm in default_global_perms:
 
        permissions[GLOBAL].add(perm.permission.permission_name)
 

	
 
    # defaults for repositories, taken from default user
 
    for perm in default_repo_perms:
 
        r_k = perm.UserRepoToPerm.repository.repo_name
 
        if perm.Repository.private and not (perm.Repository.user_id == user_id):
 
            # disable defaults for private repos,
 
            p = 'repository.none'
 
        elif perm.Repository.user_id == user_id:
 
            # set admin if owner
 
            p = 'repository.admin'
 
        else:
 
            p = perm.Permission.permission_name
 

	
 
        permissions[RK][r_k] = p
 

	
 
    # defaults for repository groups taken from default user permission
 
    # on given group
 
    for perm in default_repo_groups_perms:
 
        rg_k = perm.UserRepoGroupToPerm.group.group_name
 
        p = perm.Permission.permission_name
 
        permissions[GK][rg_k] = p
 

	
 
    # defaults for user groups taken from default user permission
 
    # on given user group
 
    for perm in default_user_group_perms:
 
        u_k = perm.UserUserGroupToPerm.user_group.users_group_name
 
        p = perm.Permission.permission_name
 
        permissions[UK][u_k] = p
 

	
 
    #======================================================================
 
    # !! OVERRIDE GLOBALS !! with user permissions if any found
 
    #======================================================================
 
    # those can be configured from groups or users explicitly
 
    _configurable = set([
 
        'hg.fork.none', 'hg.fork.repository',
 
        'hg.create.none', 'hg.create.repository',
 
        'hg.usergroup.create.false', 'hg.usergroup.create.true'
 
    ])
 

	
 
    # USER GROUPS comes first
 
    # user group global permissions
 
    user_perms_from_users_groups = Session().query(UserGroupToPerm) \
 
        .options(joinedload(UserGroupToPerm.permission)) \
 
        .join((UserGroupMember, UserGroupToPerm.users_group_id ==
 
               UserGroupMember.users_group_id)) \
 
        .filter(UserGroupMember.user_id == user_id) \
 
        .join((UserGroup, UserGroupMember.users_group_id ==
 
               UserGroup.users_group_id)) \
 
        .filter(UserGroup.users_group_active == True) \
 
        .order_by(UserGroupToPerm.users_group_id) \
 
        .all()
 
    # need to group here by groups since user can be in more than
 
    # one group
 
    _grouped = [[x, list(y)] for x, y in
 
                itertools.groupby(user_perms_from_users_groups,
 
                                  lambda x:x.users_group)]
 
    for gr, perms in _grouped:
 
        # since user can be in multiple groups iterate over them and
 
        # select the lowest permissions first (more explicit)
 
        ##TODO: do this^^
 
        if not gr.inherit_default_permissions:
 
            # NEED TO IGNORE all configurable permissions and
 
            # replace them with explicitly set
 
            permissions[GLOBAL] = permissions[GLOBAL] \
 
                                            .difference(_configurable)
 
        for perm in perms:
 
            permissions[GLOBAL].add(perm.permission.permission_name)
 

	
 
    # user specific global permissions
 
    user_perms = Session().query(UserToPerm) \
 
            .options(joinedload(UserToPerm.permission)) \
 
            .filter(UserToPerm.user_id == user_id).all()
 

	
 
    if not user_inherit_default_permissions:
 
        # NEED TO IGNORE all configurable permissions and
 
        # replace them with explicitly set
 
        permissions[GLOBAL] = permissions[GLOBAL] \
 
                                        .difference(_configurable)
 

	
 
        for perm in user_perms:
 
            permissions[GLOBAL].add(perm.permission.permission_name)
 
    ## END GLOBAL PERMISSIONS
 

	
 
    #======================================================================
 
    # !! PERMISSIONS FOR REPOSITORIES !!
 
    #======================================================================
 
    #======================================================================
 
    # check if user is part of user groups for this repository and
 
    # fill in his permission from it. _choose_perm decides of which
 
    # permission should be selected based on selected method
 
    #======================================================================
 

	
 
    # user group for repositories permissions
 
    user_repo_perms_from_users_groups = \
 
     Session().query(UserGroupRepoToPerm, Permission, Repository,) \
 
        .join((Repository, UserGroupRepoToPerm.repository_id ==
 
               Repository.repo_id)) \
 
        .join((Permission, UserGroupRepoToPerm.permission_id ==
 
               Permission.permission_id)) \
 
        .join((UserGroup, UserGroupRepoToPerm.users_group_id ==
 
               UserGroup.users_group_id)) \
 
        .filter(UserGroup.users_group_active == True) \
 
        .join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
 
               UserGroupMember.users_group_id)) \
 
        .filter(UserGroupMember.user_id == user_id) \
 
        .all()
 

	
 
    multiple_counter = collections.defaultdict(int)
 
    for perm in user_repo_perms_from_users_groups:
 
        r_k = perm.UserGroupRepoToPerm.repository.repo_name
 
        multiple_counter[r_k] += 1
 
        p = perm.Permission.permission_name
 
        cur_perm = permissions[RK][r_k]
 

	
 
        if perm.Repository.user_id == user_id:
 
            # set admin if owner
 
            p = 'repository.admin'
 
        else:
 
            if multiple_counter[r_k] > 1:
 
                p = _choose_perm(p, cur_perm)
 
        permissions[RK][r_k] = p
 

	
 
    # user explicit permissions for repositories, overrides any specified
 
    # by the group permission
 
    user_repo_perms = Permission.get_default_perms(user_id)
 
    for perm in user_repo_perms:
 
        r_k = perm.UserRepoToPerm.repository.repo_name
 
        cur_perm = permissions[RK][r_k]
 
        # set admin if owner
 
        if perm.Repository.user_id == user_id:
 
            p = 'repository.admin'
 
        else:
 
            p = perm.Permission.permission_name
 
            if not explicit:
 
                p = _choose_perm(p, cur_perm)
 
        permissions[RK][r_k] = p
 

	
 
    #======================================================================
 
    # !! PERMISSIONS FOR REPOSITORY GROUPS !!
 
    #======================================================================
 
    #======================================================================
 
    # check if user is part of user groups for this repository groups and
 
    # fill in his permission from it. _choose_perm decides of which
 
    # permission should be selected based on selected method
 
    #======================================================================
 
    # user group for repo groups permissions
 
    user_repo_group_perms_from_users_groups = \
 
     Session().query(UserGroupRepoGroupToPerm, Permission, RepoGroup) \
 
     .join((RepoGroup, UserGroupRepoGroupToPerm.group_id == RepoGroup.group_id)) \
 
     .join((Permission, UserGroupRepoGroupToPerm.permission_id
 
            == Permission.permission_id)) \
 
     .join((UserGroup, UserGroupRepoGroupToPerm.users_group_id ==
 
            UserGroup.users_group_id)) \
 
     .filter(UserGroup.users_group_active == True) \
 
     .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
 
            == UserGroupMember.users_group_id)) \
 
     .filter(UserGroupMember.user_id == user_id) \
 
     .all()
 

	
 
    multiple_counter = collections.defaultdict(int)
 
    for perm in user_repo_group_perms_from_users_groups:
 
        g_k = perm.UserGroupRepoGroupToPerm.group.group_name
 
        multiple_counter[g_k] += 1
 
        p = perm.Permission.permission_name
 
        cur_perm = permissions[GK][g_k]
 
        if multiple_counter[g_k] > 1:
 
            p = _choose_perm(p, cur_perm)
 
        permissions[GK][g_k] = p
 

	
 
    # user explicit permissions for repository groups
 
    user_repo_groups_perms = Permission.get_default_group_perms(user_id)
 
    for perm in user_repo_groups_perms:
 
        rg_k = perm.UserRepoGroupToPerm.group.group_name
 
        p = perm.Permission.permission_name
 
        cur_perm = permissions[GK][rg_k]
 
        if not explicit:
 
            p = _choose_perm(p, cur_perm)
 
        permissions[GK][rg_k] = p
 

	
 
    #======================================================================
 
    # !! PERMISSIONS FOR USER GROUPS !!
 
    #======================================================================
 
    # user group for user group permissions
 
    user_group_user_groups_perms = \
 
     Session().query(UserGroupUserGroupToPerm, Permission, UserGroup) \
 
     .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
 
            == UserGroup.users_group_id)) \
 
     .join((Permission, UserGroupUserGroupToPerm.permission_id
 
            == Permission.permission_id)) \
 
     .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
 
            == UserGroupMember.users_group_id)) \
 
     .filter(UserGroupMember.user_id == user_id) \
 
     .join((UserGroup, UserGroupMember.users_group_id ==
 
            UserGroup.users_group_id), aliased=True, from_joinpoint=True) \
 
     .filter(UserGroup.users_group_active == True) \
 
     .all()
 

	
 
    multiple_counter = collections.defaultdict(int)
 
    for perm in user_group_user_groups_perms:
 
        g_k = perm.UserGroupUserGroupToPerm.target_user_group.users_group_name
 
        multiple_counter[g_k] += 1
 
        p = perm.Permission.permission_name
 
        cur_perm = permissions[UK][g_k]
 
        if multiple_counter[g_k] > 1:
 
            p = _choose_perm(p, cur_perm)
 
        permissions[UK][g_k] = p
 

	
 
    #user explicit permission for user groups
 
    user_user_groups_perms = Permission.get_default_user_group_perms(user_id)
 
    for perm in user_user_groups_perms:
 
        u_k = perm.UserUserGroupToPerm.user_group.users_group_name
 
        p = perm.Permission.permission_name
 
        cur_perm = permissions[UK][u_k]
 
        if not explicit:
 
            p = _choose_perm(p, cur_perm)
 
        permissions[UK][u_k] = p
 

	
 
    return permissions
 

	
 

	
 
def allowed_api_access(controller_name, whitelist=None, api_key=None):
 
    """
 
    Check if given controller_name is in whitelist API access
 
    """
 
    if not whitelist:
 
        from kallithea import CONFIG
 
        whitelist = aslist(CONFIG.get('api_access_controllers_whitelist'),
 
                           sep=',')
 
        log.debug('whitelist of API access is: %s', whitelist)
 
    api_access_valid = controller_name in whitelist
 
    if api_access_valid:
 
        log.debug('controller:%s is in API whitelist', controller_name)
 
    else:
 
        msg = 'controller: %s is *NOT* in API whitelist' % (controller_name)
 
        if api_key:
 
            #if we use API key and don't have access it's a warning
 
            log.warning(msg)
 
        else:
 
            log.debug(msg)
 
    return api_access_valid
 

	
 

	
 
class AuthUser(object):
 
    """
 
    Represents a Kallithea user, including various authentication and
 
    authorization information. Typically used to store the current user,
 
    but is also used as a generic user information data structure in
 
    parts of the code, e.g. user management.
 

	
 
    Constructed from a database `User` object, a user ID or cookie dict,
 
    it looks up the user (if needed) and copies all attributes to itself,
 
    adding various non-persistent data. If lookup fails but anonymous
 
    access to Kallithea is enabled, the default user is loaded instead.
 

	
 
    `AuthUser` does not by itself authenticate users and the constructor
 
    sets the `is_authenticated` field to False. It's up to other parts
 
    of the code to check e.g. if a supplied password is correct, and if
 
    so, set `is_authenticated` to True.
 

	
 
    However, `AuthUser` does refuse to load a user that is not `active`.
 
    """
 

	
 
    def __init__(self, user_id=None, dbuser=None,
 
            is_external_auth=False):
 

	
 
        self.is_authenticated = False
 
        self.is_external_auth = is_external_auth
 

	
 
        user_model = UserModel()
 
        self.anonymous_user = User.get_default_user(cache=True)
 

	
 
        # These attributes will be overriden by fill_data, below, unless the
 
        # These attributes will be overridden by fill_data, below, unless the
 
        # requested user cannot be found and the default anonymous user is
 
        # not enabled.
 
        self.user_id = None
 
        self.username = None
 
        self.api_key = None
 
        self.name = ''
 
        self.lastname = ''
 
        self.email = ''
 
        self.admin = False
 
        self.inherit_default_permissions = False
 

	
 
        # Look up database user, if necessary.
 
        if user_id is not None:
 
            log.debug('Auth User lookup by USER ID %s', user_id)
 
            dbuser = user_model.get(user_id)
 
        else:
 
            # Note: dbuser is allowed to be None.
 
            log.debug('Auth User lookup by database user %s', dbuser)
 

	
 
        is_user_loaded = self._fill_data(dbuser)
 

	
 
        # If user cannot be found, try falling back to anonymous.
 
        if not is_user_loaded:
 
            is_user_loaded =  self._fill_data(self.anonymous_user)
 

	
 
        self.is_default_user = (self.user_id == self.anonymous_user.user_id)
 

	
 
        if not self.username:
 
            self.username = 'None'
 

	
 
        log.debug('Auth User is now %s', self)
 

	
 
    def _fill_data(self, dbuser):
 
        """
 
        Copies database fields from a `db.User` to this `AuthUser`. Does
 
        not copy `api_keys` and `permissions` attributes.
 

	
 
        Checks that `dbuser` is `active` (and not None) before copying;
 
        returns True on success.
 
        """
 
        if dbuser is not None and dbuser.active:
 
            log.debug('filling %s data', dbuser)
 
            for k, v in dbuser.get_dict().iteritems():
 
                assert k not in ['api_keys', 'permissions']
 
                setattr(self, k, v)
 
            return True
 
        return False
 

	
 
    @LazyProperty
 
    def permissions(self):
 
        return self.__get_perms(user=self, cache=False)
 

	
 
    @property
 
    def api_keys(self):
 
        return self._get_api_keys()
 

	
 
    def __get_perms(self, user, explicit=True, algo='higherwin', cache=False):
 
        """
 
        Fills user permission attribute with permissions taken from database
 
        works for permissions given for repositories, and for permissions that
 
        are granted to groups
 

	
 
        :param user: `AuthUser` instance
 
        :param explicit: In case there are permissions both for user and a group
 
            that user is part of, explicit flag will define if user will
 
            explicitly override permissions from group, if it's False it will
 
            make decision based on the algo
 
        :param algo: algorithm to decide what permission should be choose if
 
            it's multiple defined, eg user in two different groups. It also
 
            decides if explicit flag is turned off how to specify the permission
 
            for case when user is in a group + have defined separate permission
 
        """
 
        user_id = user.user_id
 
        user_is_admin = user.is_admin
 
        user_inherit_default_permissions = user.inherit_default_permissions
 

	
 
        log.debug('Getting PERMISSION tree')
 
        compute = conditional_cache('short_term', 'cache_desc',
 
                                    condition=cache, func=_cached_perms_data)
 
        return compute(user_id, user_is_admin,
 
                       user_inherit_default_permissions, explicit, algo)
 

	
 
    def _get_api_keys(self):
 
        api_keys = [self.api_key]
 
        for api_key in UserApiKeys.query() \
 
                .filter(UserApiKeys.user_id == self.user_id) \
 
                .filter(or_(UserApiKeys.expires == -1,
 
                            UserApiKeys.expires >= time.time())).all():
 
            api_keys.append(api_key.api_key)
 

	
 
        return api_keys
 

	
 
    @property
 
    def is_admin(self):
 
        return self.admin
 

	
 
    @property
 
    def repositories_admin(self):
 
        """
 
        Returns list of repositories you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['repositories'].iteritems()
 
                if x[1] == 'repository.admin']
 

	
 
    @property
 
    def repository_groups_admin(self):
 
        """
 
        Returns list of repository groups you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['repositories_groups'].iteritems()
 
                if x[1] == 'group.admin']
 

	
 
    @property
 
    def user_groups_admin(self):
 
        """
 
        Returns list of user groups you're an admin of
 
        """
 
        return [x[0] for x in self.permissions['user_groups'].iteritems()
 
                if x[1] == 'usergroup.admin']
 

	
 
    @staticmethod
 
    def check_ip_allowed(user, ip_addr):
 
        """
 
        Check if the given IP address (a `str`) is allowed for the given
 
        user (an `AuthUser` or `db.User`).
 
        """
 
        allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,
 
            inherit_from_default=user.inherit_default_permissions)
 
        if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):
 
            log.debug('IP:%s is in range of %s', ip_addr, allowed_ips)
 
            return True
 
        else:
 
            log.info('Access for IP:%s forbidden, '
 
                     'not in %s' % (ip_addr, allowed_ips))
 
            return False
 

	
 
    def __repr__(self):
 
        return "<AuthUser('id:%s[%s] auth:%s')>" \
 
            % (self.user_id, self.username, (self.is_authenticated or self.is_default_user))
 

	
 
    def to_cookie(self):
 
        """ Serializes this login session to a cookie `dict`. """
 
        return {
 
            'user_id': self.user_id,
 
            'is_external_auth': self.is_external_auth,
 
        }
 

	
 
    @staticmethod
 
    def from_cookie(cookie):
 
        """
 
        Deserializes an `AuthUser` from a cookie `dict`.
 
        """
 

	
 
        au = AuthUser(
 
            user_id=cookie.get('user_id'),
 
            is_external_auth=cookie.get('is_external_auth', False),
 
        )
 
        au.is_authenticated = True
 
        return au
 

	
 
    @classmethod
 
    def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):
 
        _set = set()
 

	
 
        if inherit_from_default:
 
            default_ips = UserIpMap.query().filter(UserIpMap.user ==
 
                                            User.get_default_user(cache=True))
 
            if cache:
 
                default_ips = default_ips.options(FromCache("sql_cache_short",
 
                                                  "get_user_ips_default"))
 

	
 
            # populate from default user
 
            for ip in default_ips:
 
                try:
 
                    _set.add(ip.ip_addr)
 
                except ObjectDeletedError:
 
                    # since we use heavy caching sometimes it happens that we get
 
                    # deleted objects here, we just skip them
 
                    pass
 

	
 
        user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
 
        if cache:
 
            user_ips = user_ips.options(FromCache("sql_cache_short",
 
                                                  "get_user_ips_%s" % user_id))
 

	
 
        for ip in user_ips:
 
            try:
 
                _set.add(ip.ip_addr)
 
            except ObjectDeletedError:
 
                # since we use heavy caching sometimes it happens that we get
 
                # deleted objects here, we just skip them
 
                pass
 
        return _set or set(['0.0.0.0/0', '::/0'])
 

	
 

	
 
def set_available_permissions(config):
 
    """
 
    This function will propagate pylons globals with all available defined
 
    permission given in db. We don't want to check each time from db for new
 
    permissions since adding a new permission also requires application restart
 
    ie. to decorate new views with the newly created permission
 

	
 
    :param config: current pylons config instance
 

	
 
    """
 
    log.info('getting information about all available permissions')
 
    try:
 
        sa = meta.Session
 
        all_perms = sa.query(Permission).all()
 
        config['available_permissions'] = [x.permission_name for x in all_perms]
 
    finally:
 
        meta.Session.remove()
 

	
 

	
 
#==============================================================================
 
# CHECK DECORATORS
 
#==============================================================================
 

	
 
def _redirect_to_login(message=None):
 
    """Return an exception that must be raised. It will redirect to the login
 
    page which will redirect back to the current URL after authentication.
 
    The optional message will be shown in a flash message."""
 
    from kallithea.lib import helpers as h
 
    if message:
 
        h.flash(h.literal(message), category='warning')
 
    p = request.path_qs
 
    log.debug('Redirecting to login page, origin: %s', p)
 
    return HTTPFound(location=url('login_home', came_from=p))
 

	
 

	
 
class LoginRequired(object):
 
    """
 
    Must be logged in to execute this function else
 
    redirect to login page
 

	
 
    :param api_access: if enabled this checks only for valid auth token
 
        and grants access based on valid token
 
    """
 

	
 
    def __init__(self, api_access=False):
 
        self.api_access = api_access
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        controller = fargs[0]
 
        user = controller.authuser
 
        loc = "%s:%s" % (controller.__class__.__name__, func.__name__)
 
        log.debug('Checking access for user %s @ %s', user, loc)
 

	
 
        if not AuthUser.check_ip_allowed(user, controller.ip_addr):
 
            raise _redirect_to_login(_('IP %s not allowed') % controller.ip_addr)
 

	
 
        # check if we used an API key and it's a valid one
 
        api_key = request.GET.get('api_key')
 
        if api_key is not None:
 
            # explicit controller is enabled or API is in our whitelist
 
            if self.api_access or allowed_api_access(loc, api_key=api_key):
 
                if api_key in user.api_keys:
 
                    log.info('user %s authenticated with API key ****%s @ %s',
 
                             user, api_key[-4:], loc)
 
                    return func(*fargs, **fkwargs)
 
                else:
 
                    log.warning('API key ****%s is NOT valid', api_key[-4:])
 
                    raise _redirect_to_login(_('Invalid API key'))
 
            else:
 
                # controller does not allow API access
 
                log.warning('API access to %s is not allowed', loc)
 
                raise HTTPForbidden()
 

	
 
        # Only allow the following HTTP request methods. (We sometimes use POST
 
        # requests with a '_method' set to 'PUT' or 'DELETE'; but that is only
 
        # used for the route lookup, and does not affect request.method.)
 
        if request.method not in ['GET', 'HEAD', 'POST', 'PUT']:
 
            raise HTTPMethodNotAllowed()
 

	
 
        # Also verify the _method override. This is only permitted in POST
 
        # requests, and can specify PUT or DELETE.
 
        _method = request.params.get('_method')
 
        if _method is None:
 
            pass # no override, no problem
 
        elif request.method == 'POST' and _method.upper() in ['PUT', 'DELETE']:
 
            pass # permitted override
 
        else:
 
            raise HTTPMethodNotAllowed()
 

	
 
        # Make sure CSRF token never appears in the URL. If so, invalidate it.
 
        if secure_form.token_key in request.GET:
 
            log.error('CSRF key leak detected')
 
            session.pop(secure_form.token_key, None)
 
            session.save()
 
            from kallithea.lib import helpers as h
 
            h.flash(_("CSRF token leak has been detected - all form tokens have been expired"),
 
                    category='error')
 

	
 
        # 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.
 
        if request.method not in ['GET', 'HEAD']:
 
            token = request.POST.get(secure_form.token_key)
 
            if not token or token != secure_form.authentication_token():
 
                log.error('CSRF check failed')
 
                raise HTTPForbidden()
 

	
 
        # 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 HTTPBadRequest()
 

	
 
        # regular user authentication
 
        if user.is_authenticated or user.is_default_user:
 
            log.info('user %s authenticated with regular auth @ %s', user, loc)
 
            return func(*fargs, **fkwargs)
 
        else:
 
            log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)
 
            raise _redirect_to_login()
 

	
 
class NotAnonymous(object):
 
    """
 
    Must be logged in to execute this function else
 
    redirect to login page"""
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        cls = fargs[0]
 
        self.user = cls.authuser
 

	
 
        log.debug('Checking if user is not anonymous @%s', cls)
 

	
 
        if self.user.is_default_user:
 
            raise _redirect_to_login(_('You need to be a registered user to '
 
                                       'perform this action'))
 
        else:
 
            return func(*fargs, **fkwargs)
 

	
 

	
 
class PermsDecorator(object):
 
    """Base class for controller decorators"""
 

	
 
    def __init__(self, *required_perms):
 
        self.required_perms = set(required_perms)
 
        self.user_perms = None
 

	
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 

	
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        cls = fargs[0]
 
        self.user = cls.authuser
 
        self.user_perms = self.user.permissions
 
        log.debug('checking %s permissions %s for %s %s',
 
          self.__class__.__name__, self.required_perms, cls, self.user)
 

	
 
        if self.check_permissions():
 
            log.debug('Permission granted for %s %s', cls, self.user)
 
            return func(*fargs, **fkwargs)
 

	
 
        else:
 
            log.debug('Permission denied for %s %s', cls, self.user)
 
            if self.user.is_default_user:
 
                raise _redirect_to_login(_('You need to be signed in to view this page'))
 
            else:
 
                raise HTTPForbidden()
 

	
 
    def check_permissions(self):
 
        """Dummy function for overriding"""
 
        raise Exception('You have to write this function in child class')
 

	
 

	
 
class HasPermissionAllDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for all given predicates. All of them
 
    have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        if self.required_perms.issubset(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 

	
 
class HasPermissionAnyDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for any of given predicates. In order to
 
    fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        if self.required_perms.intersection(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoPermissionAllDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for all given predicates for specific
 
    repository. All of them have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        repo_name = get_repo_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['repositories'][repo_name]])
 
        except KeyError:
 
            return False
 
        if self.required_perms.issubset(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoPermissionAnyDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for any of given predicates for specific
 
    repository. In order to fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        repo_name = get_repo_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['repositories'][repo_name]])
 
        except KeyError:
 
            return False
 

	
 
        if self.required_perms.intersection(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoGroupPermissionAllDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for all given predicates for specific
 
    repository group. All of them have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        group_name = get_repo_group_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['repositories_groups'][group_name]])
 
        except KeyError:
 
            return False
 

	
 
        if self.required_perms.issubset(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoGroupPermissionAnyDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for any of given predicates for specific
 
    repository group. In order to fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        group_name = get_repo_group_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['repositories_groups'][group_name]])
 
        except KeyError:
 
            return False
 

	
 
        if self.required_perms.intersection(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasUserGroupPermissionAllDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for all given predicates for specific
 
    user group. All of them have to be meet in order to fulfill the request
 
    """
 

	
 
    def check_permissions(self):
 
        group_name = get_user_group_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['user_groups'][group_name]])
 
        except KeyError:
 
            return False
 

	
 
        if self.required_perms.issubset(user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasUserGroupPermissionAnyDecorator(PermsDecorator):
 
    """
 
    Checks for access permission for any of given predicates for specific
 
    user group. In order to fulfill the request any of predicates must be meet
 
    """
 

	
 
    def check_permissions(self):
 
        group_name = get_user_group_slug(request)
 
        try:
 
            user_perms = set([self.user_perms['user_groups'][group_name]])
 
        except KeyError:
 
            return False
 

	
 
        if self.required_perms.intersection(user_perms):
 
            return True
 
        return False
 

	
 

	
 
#==============================================================================
 
# CHECK FUNCTIONS
 
#==============================================================================
 
class PermsFunction(object):
 
    """Base function for other check functions"""
 

	
 
    def __init__(self, *perms):
 
        self.required_perms = set(perms)
 
        self.user_perms = None
 
        self.repo_name = None
 
        self.group_name = None
 

	
 
    def __call__(self, check_location='', user=None):
 
        if not user:
 
            #TODO: remove this someday,put as user as attribute here
 
            user = request.user
 

	
 
        # init auth user if not already given
 
        if not isinstance(user, AuthUser):
 
            user = AuthUser(user.user_id)
 

	
 
        cls_name = self.__class__.__name__
 
        check_scope = {
 
            'HasPermissionAll': '',
 
            'HasPermissionAny': '',
 
            'HasRepoPermissionAll': 'repo:%s' % self.repo_name,
 
            'HasRepoPermissionAny': 'repo:%s' % self.repo_name,
 
            'HasRepoGroupPermissionAll': 'group:%s' % self.group_name,
 
            'HasRepoGroupPermissionAny': 'group:%s' % self.group_name,
 
        }.get(cls_name, '?')
 
        log.debug('checking cls:%s %s usr:%s %s @ %s', cls_name,
 
                  self.required_perms, user, check_scope,
 
                  check_location or 'unspecified location')
 
        if not user:
 
            log.debug('Empty request user')
 
            return False
 
        self.user_perms = user.permissions
 
        if self.check_permissions():
 
            log.debug('Permission to %s granted for user: %s @ %s',
 
                      check_scope, user,
 
                         check_location or 'unspecified location')
 
            return True
 

	
 
        else:
 
            log.debug('Permission to %s denied for user: %s @ %s',
 
                      check_scope, user,
 
                         check_location or 'unspecified location')
 
            return False
 

	
 
    def check_permissions(self):
 
        """Dummy function for overriding"""
 
        raise Exception('You have to write this function in child class')
 

	
 

	
 
class HasPermissionAll(PermsFunction):
 
    def check_permissions(self):
 
        if self.required_perms.issubset(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 

	
 
class HasPermissionAny(PermsFunction):
 
    def check_permissions(self):
 
        if self.required_perms.intersection(self.user_perms.get('global')):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoPermissionAll(PermsFunction):
 
    def __call__(self, repo_name=None, check_location='', user=None):
 
        self.repo_name = repo_name
 
        return super(HasRepoPermissionAll, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        if not self.repo_name:
 
            self.repo_name = get_repo_slug(request)
 

	
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['repositories'][self.repo_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.issubset(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoPermissionAny(PermsFunction):
 
    def __call__(self, repo_name=None, check_location='', user=None):
 
        self.repo_name = repo_name
 
        return super(HasRepoPermissionAny, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        if not self.repo_name:
 
            self.repo_name = get_repo_slug(request)
 

	
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['repositories'][self.repo_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.intersection(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoGroupPermissionAny(PermsFunction):
 
    def __call__(self, group_name=None, check_location='', user=None):
 
        self.group_name = group_name
 
        return super(HasRepoGroupPermissionAny, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['repositories_groups'][self.group_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.intersection(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasRepoGroupPermissionAll(PermsFunction):
 
    def __call__(self, group_name=None, check_location='', user=None):
 
        self.group_name = group_name
 
        return super(HasRepoGroupPermissionAll, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['repositories_groups'][self.group_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.issubset(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasUserGroupPermissionAny(PermsFunction):
 
    def __call__(self, user_group_name=None, check_location='', user=None):
 
        self.user_group_name = user_group_name
 
        return super(HasUserGroupPermissionAny, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['user_groups'][self.user_group_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.intersection(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
class HasUserGroupPermissionAll(PermsFunction):
 
    def __call__(self, user_group_name=None, check_location='', user=None):
 
        self.user_group_name = user_group_name
 
        return super(HasUserGroupPermissionAll, self).__call__(check_location, user)
 

	
 
    def check_permissions(self):
 
        try:
 
            self._user_perms = set(
 
                [self.user_perms['user_groups'][self.user_group_name]]
 
            )
 
        except KeyError:
 
            return False
 
        if self.required_perms.issubset(self._user_perms):
 
            return True
 
        return False
 

	
 

	
 
#==============================================================================
 
# SPECIAL VERSION TO HANDLE MIDDLEWARE AUTH
 
#==============================================================================
 
class HasPermissionAnyMiddleware(object):
 
    def __init__(self, *perms):
 
        self.required_perms = set(perms)
 

	
 
    def __call__(self, user, repo_name):
 
        # repo_name MUST be unicode, since we handle keys in permission
 
        # dict by unicode
 
        repo_name = safe_unicode(repo_name)
 
        usr = AuthUser(user.user_id)
 
        self.user_perms = set([usr.permissions['repositories'][repo_name]])
 
        self.username = user.username
 
        self.repo_name = repo_name
 
        return self.check_permissions()
 

	
 
    def check_permissions(self):
 
        log.debug('checking VCS protocol '
 
                  'permissions %s for user:%s repository:%s', self.user_perms,
 
                                                self.username, self.repo_name)
 
        if self.required_perms.intersection(self.user_perms):
 
            log.debug('Permission to repo: %s granted for user: %s @ %s',
 
                      self.repo_name, self.username, 'PermissionMiddleware')
 
            return True
 
        log.debug('Permission to repo: %s denied for user: %s @ %s',
 
                  self.repo_name, self.username, 'PermissionMiddleware')
 
        return False
 

	
 

	
 
#==============================================================================
 
# SPECIAL VERSION TO HANDLE API AUTH
 
#==============================================================================
 
class _BaseApiPerm(object):
 
    def __init__(self, *perms):
 
        self.required_perms = set(perms)
 

	
 
    def __call__(self, check_location=None, user=None, repo_name=None,
 
                 group_name=None):
 
        cls_name = self.__class__.__name__
 
        check_scope = 'user:%s' % (user)
 
        if repo_name:
 
            check_scope += ', repo:%s' % (repo_name)
 

	
 
        if group_name:
 
            check_scope += ', repo group:%s' % (group_name)
 

	
 
        log.debug('checking cls:%s %s %s @ %s',
 
                  cls_name, self.required_perms, check_scope, check_location)
 
        if not user:
 
            log.debug('Empty User passed into arguments')
 
            return False
 

	
 
        ## process user
 
        if not isinstance(user, AuthUser):
 
            user = AuthUser(user.user_id)
 
        if not check_location:
 
            check_location = 'unspecified'
 
        if self.check_permissions(user.permissions, repo_name, group_name):
 
            log.debug('Permission to %s granted for user: %s @ %s',
 
                      check_scope, user, check_location)
 
            return True
 

	
 
        else:
 
            log.debug('Permission to %s denied for user: %s @ %s',
 
                      check_scope, user, check_location)
 
            return False
 

	
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):
 
        """
 
        implement in child class should return True if permissions are ok,
 
        False otherwise
 

	
 
        :param perm_defs: dict with permission definitions
 
        :param repo_name: repo name
 
        """
 
        raise NotImplementedError()
 

	
 

	
 
class HasPermissionAllApi(_BaseApiPerm):
 
    def check_permissions(self, perm_defs, repo_name=None, group_name=None):
 
        if self.required_perms.issubset(perm_defs.get('global')):
 
            return True
 
        return False
kallithea/tests/functional/test_login.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import re
 
import time
 
import urlparse
 

	
 
import mock
 

	
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib.utils2 import generate_api_key
 
from kallithea.lib.auth import check_password
 
from kallithea.lib import helpers as h
 
from kallithea.model.api_key import ApiKeyModel
 
from kallithea.model import validators
 
from kallithea.model.db import User, Notification
 
from kallithea.model.meta import Session
 
from kallithea.model.user import UserModel
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestLoginController(TestControllerPytest):
 
    def setup_method(self, method):
 
        remove_all_notifications()
 
        self.assertEqual(Notification.query().all(), [])
 

	
 
    def test_index(self):
 
        response = self.app.get(url(controller='login', action='index'))
 
        self.assertEqual(response.status, '200 OK')
 
        # Test response...
 

	
 
    def test_login_admin_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
        self.assertEqual(response.status, '302 Found')
 
        self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS})
 

	
 
        self.assertEqual(response.status, '302 Found')
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_email_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_EMAIL,
 
                                  'password': TEST_USER_REGULAR_PASS})
 

	
 
        self.assertEqual(response.status, '302 Found')
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_ok_came_from(self):
 
        test_came_from = '/_admin/users'
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from=test_came_from),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
        self.assertEqual(response.status, '302 Found')
 
        response = response.follow()
 

	
 
        self.assertEqual(response.status, '200 OK')
 
        response.mustcontain('Users Administration')
 

	
 
    def test_login_do_not_remember(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS,
 
                                  'remember': False})
 

	
 
        self.assertIn('Set-Cookie', response.headers)
 
        for cookie in response.headers.getall('Set-Cookie'):
 
            self.assertFalse(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
 
                'Cookie %r has expiration date, but should be a session cookie' % cookie)
 

	
 
    def test_login_remember(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS,
 
                                  'remember': True})
 

	
 
        self.assertIn('Set-Cookie', response.headers)
 
        for cookie in response.headers.getall('Set-Cookie'):
 
            self.assertTrue(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
 
                'Cookie %r should have expiration date, but is a session cookie' % cookie)
 

	
 
    def test_logout(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS})
 

	
 
        # Verify that a login session has been established.
 
        response = self.app.get(url(controller='login', action='index'))
 
        response = response.follow()
 
        self.assertIn('authuser', response.session)
 

	
 
        response.click('Log Out')
 

	
 
        # Verify that the login session has been terminated.
 
        response = self.app.get(url(controller='login', action='index'))
 
        self.assertNotIn('authuser', response.session)
 

	
 
    @parametrize('url_came_from', [
 
          ('data:text/html,<script>window.alert("xss")</script>',),
 
          ('mailto:test@example.com',),
 
          ('file:///etc/passwd',),
 
          ('ftp://ftp.example.com',),
 
          ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
 
          ('//evil.example.com/',),
 
          ('/\r\nX-Header-Injection: boo',),
 
          ('/invälid_url_bytes',),
 
          ('non-absolute-path',),
 
    ])
 
    def test_login_bad_came_froms(self, url_came_from):
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from=url_came_from),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS},
 
                                 status=400)
 

	
 
    def test_login_short_password(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': 'as'})
 
        self.assertEqual(response.status, '200 OK')
 

	
 
        response.mustcontain('Enter 3 characters or more')
 

	
 
    def test_login_wrong_username_password(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': 'error',
 
                                  'password': 'test12'})
 

	
 
        response.mustcontain('Invalid username or password')
 

	
 
    # verify that get arguments are correctly passed along login redirection
 

	
 
    @parametrize('args,args_encoded', [
 
        ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),
 
        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
 
             (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),
 
    ])
 
    def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
 
        with fixture.anon_access(False):
 
            response = self.app.get(url(controller='summary', action='index',
 
                                        repo_name=HG_REPO,
 
                                        **args))
 
            self.assertEqual(response.status, '302 Found')
 
            came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]
 
            came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)
 
            for encoded in args_encoded:
 
                self.assertIn(encoded, came_from_qs)
 

	
 
    @parametrize('args,args_encoded', [
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå', 'green':u'grøn'},
 
             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
 
    ])
 
    def test_login_form_preserves_get_args(self, args, args_encoded):
 
        response = self.app.get(url(controller='login', action='index',
 
                                    came_from=url('/_admin/users', **args)))
 
        came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, came_from)
 

	
 
    @parametrize('args,args_encoded', [
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå', 'green':u'grøn'},
 
             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
 
    ])
 
    def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from = url('/_admin/users', **args)),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
        self.assertEqual(response.status, '302 Found')
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, response.location)
 

	
 
    @parametrize('args,args_encoded', [
 
        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
 
        ({'blue': u'blå', 'green':u'grøn'},
 
             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
 
    ])
 
    def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from=url('/_admin/users', **args)),
 
                                 {'username': 'error',
 
                                  'password': 'test12'})
 

	
 
        response.mustcontain('Invalid username or password')
 
        came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
 
        for encoded in args_encoded:
 
            self.assertIn(encoded, came_from)
 

	
 
    #==========================================================================
 
    # REGISTRATIONS
 
    #==========================================================================
 
    def test_register(self):
 
        response = self.app.get(url(controller='login', action='register'))
 
        response.mustcontain('Sign Up')
 

	
 
    def test_register_err_same_username(self):
 
        uname = TEST_USER_ADMIN_LOGIN
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': uname,
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmail@example.com',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        msg = validators.ValidUsername()._messages['username_exists']
 
        msg = h.html_escape(msg % {'username': uname})
 
        response.mustcontain(msg)
 

	
 
    def test_register_err_same_email(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'test_admin_0',
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': TEST_USER_ADMIN_EMAIL,
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        msg = validators.UniqSystemEmail()()._messages['email_taken']
 
        response.mustcontain(msg)
 

	
 
    def test_register_err_same_email_case_sensitive(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'test_admin_1',
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': TEST_USER_ADMIN_EMAIL.title(),
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 
        msg = validators.UniqSystemEmail()()._messages['email_taken']
 
        response.mustcontain(msg)
 

	
 
    def test_register_err_wrong_data(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'xs',
 
                                             'password': 'test',
 
                                             'password_confirmation': 'test',
 
                                             'email': 'goodmailm',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 
        self.assertEqual(response.status, '200 OK')
 
        response.mustcontain('An email address must contain a single @')
 
        response.mustcontain('Enter a value 6 characters long or more')
 

	
 
    def test_register_err_username(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'error user',
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmailm',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        response.mustcontain('An email address must contain a single @')
 
        response.mustcontain('Username may only contain '
 
                'alphanumeric characters underscores, '
 
                'periods or dashes and must begin with an '
 
                'alphanumeric character')
 

	
 
    def test_register_err_case_sensitive(self):
 
        usr = TEST_USER_ADMIN_LOGIN.title()
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': usr,
 
                                             'password': 'test12',
 
                                             'password_confirmation': 'test12',
 
                                             'email': 'goodmailm',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 

	
 
        response.mustcontain('An email address must contain a single @')
 
        msg = validators.ValidUsername()._messages['username_exists']
 
        msg = h.html_escape(msg % {'username': usr})
 
        response.mustcontain(msg)
 

	
 
    def test_register_special_chars(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                        {'username': 'xxxaxn',
 
                                         'password': 'ąćźżąśśśś',
 
                                         'password_confirmation': 'ąćźżąśśśś',
 
                                         'email': 'goodmailm@test.plx',
 
                                         'firstname': 'test',
 
                                         'lastname': 'test'})
 

	
 
        msg = validators.ValidPassword()._messages['invalid_password']
 
        response.mustcontain(msg)
 

	
 
    def test_register_password_mismatch(self):
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': 'xs',
 
                                             'password': '123qwe',
 
                                             'password_confirmation': 'qwe123',
 
                                             'email': 'goodmailm@test.plxa',
 
                                             'firstname': 'test',
 
                                             'lastname': 'test'})
 
        msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
 
        response.mustcontain(msg)
 

	
 
    def test_register_ok(self):
 
        username = 'test_regular4'
 
        password = 'qweqwe'
 
        email = 'user4@example.com'
 
        name = 'testname'
 
        lastname = 'testlastname'
 

	
 
        response = self.app.post(url(controller='login', action='register'),
 
                                            {'username': username,
 
                                             'password': password,
 
                                             'password_confirmation': password,
 
                                             'email': email,
 
                                             'firstname': name,
 
                                             'lastname': lastname,
 
                                             'admin': True})  # This should be overriden
 
                                             'admin': True})  # This should be overridden
 
        self.assertEqual(response.status, '302 Found')
 
        self.checkSessionFlash(response, 'You have successfully registered into Kallithea')
 

	
 
        ret = Session().query(User).filter(User.username == 'test_regular4').one()
 
        self.assertEqual(ret.username, username)
 
        self.assertEqual(check_password(password, ret.password), True)
 
        self.assertEqual(ret.email, email)
 
        self.assertEqual(ret.name, name)
 
        self.assertEqual(ret.lastname, lastname)
 
        self.assertNotEqual(ret.api_key, None)
 
        self.assertEqual(ret.admin, False)
 

	
 
    #==========================================================================
 
    # PASSWORD RESET
 
    #==========================================================================
 

	
 
    def test_forgot_password_wrong_mail(self):
 
        bad_email = 'username%wrongmail.org'
 
        response = self.app.post(
 
                        url(controller='login', action='password_reset'),
 
                            {'email': bad_email, }
 
        )
 

	
 
        response.mustcontain('An email address must contain a single @')
 

	
 
    def test_forgot_password(self):
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset'))
 
        self.assertEqual(response.status, '200 OK')
 

	
 
        username = 'test_password_reset_1'
 
        password = 'qweqwe'
 
        email = 'username@example.com'
 
        name = u'passwd'
 
        lastname = u'reset'
 
        timestamp = int(time.time())
 

	
 
        new = User()
 
        new.username = username
 
        new.password = password
 
        new.email = email
 
        new.name = name
 
        new.lastname = lastname
 
        new.api_key = generate_api_key()
 
        Session().add(new)
 
        Session().commit()
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset'),
 
                                 {'email': email, })
 

	
 
        self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
 

	
 
        response = response.follow()
 

	
 
        # BAD TOKEN
 

	
 
        token = "bad"
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset_confirmation'),
 
                                 {'email': email,
 
                                  'timestamp': timestamp,
 
                                  'password': "p@ssw0rd",
 
                                  'password_confirm': "p@ssw0rd",
 
                                  'token': token,
 
                                 })
 
        self.assertEqual(response.status, '200 OK')
 
        response.mustcontain('Invalid password reset token')
 

	
 
        # GOOD TOKEN
 

	
 
        # TODO: The token should ideally be taken from the mail sent
 
        # above, instead of being recalculated.
 

	
 
        token = UserModel().get_reset_password_token(
 
            User.get_by_username(username), timestamp, self.authentication_token())
 

	
 
        response = self.app.get(url(controller='login',
 
                                    action='password_reset_confirmation',
 
                                    email=email,
 
                                    timestamp=timestamp,
 
                                    token=token))
 
        self.assertEqual(response.status, '200 OK')
 
        response.mustcontain("You are about to set a new password for the email address %s" % email)
 

	
 
        response = self.app.post(url(controller='login',
 
                                     action='password_reset_confirmation'),
 
                                 {'email': email,
 
                                  'timestamp': timestamp,
 
                                  'password': "p@ssw0rd",
 
                                  'password_confirm': "p@ssw0rd",
 
                                  'token': token,
 
                                 })
 
        self.assertEqual(response.status, '302 Found')
 
        self.checkSessionFlash(response, 'Successfully updated password')
 

	
 
        response = response.follow()
 

	
 
    #==========================================================================
 
    # API
 
    #==========================================================================
 

	
 
    def _get_api_whitelist(self, values=None):
 
        config = {'api_access_controllers_whitelist': values or []}
 
        return config
 

	
 
    @parametrize('test_name,api_key', [
 
        ('none', None),
 
        ('empty_string', ''),
 
        ('fake_number', '123456'),
 
        ('proper_api_key', None)
 
    ])
 
    def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key):
 
        whitelist = self._get_api_whitelist([])
 
        with mock.patch('kallithea.CONFIG', whitelist):
 
            self.assertEqual([],
 
                             whitelist['api_access_controllers_whitelist'])
 
            if test_name == 'proper_api_key':
 
                #use builtin if api_key is None
 
                api_key = User.get_first_admin().api_key
 

	
 
            with fixture.anon_access(False):
 
                self.app.get(url(controller='changeset',
 
                                 action='changeset_raw',
 
                                 repo_name=HG_REPO, revision='tip', api_key=api_key),
 
                             status=403)
 

	
 
    @parametrize('test_name,api_key,code', [
 
        ('none', None, 302),
 
        ('empty_string', '', 302),
 
        ('fake_number', '123456', 302),
 
        ('fake_not_alnum', 'a-z', 302),
 
        ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 302),
 
        ('proper_api_key', None, 200)
 
    ])
 
    def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):
 
        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
 
        with mock.patch('kallithea.CONFIG', whitelist):
 
            self.assertEqual(['ChangesetController:changeset_raw'],
 
                             whitelist['api_access_controllers_whitelist'])
 
            if test_name == 'proper_api_key':
 
                api_key = User.get_first_admin().api_key
 

	
 
            with fixture.anon_access(False):
 
                self.app.get(url(controller='changeset',
 
                                 action='changeset_raw',
 
                                 repo_name=HG_REPO, revision='tip', api_key=api_key),
 
                             status=code)
 

	
 
    def test_access_page_via_extra_api_key(self):
 
        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
 
        with mock.patch('kallithea.CONFIG', whitelist):
 
            self.assertEqual(['ChangesetController:changeset_raw'],
 
                             whitelist['api_access_controllers_whitelist'])
 

	
 
            new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
 
            Session().commit()
 
            with fixture.anon_access(False):
 
                self.app.get(url(controller='changeset',
 
                                 action='changeset_raw',
 
                                 repo_name=HG_REPO, revision='tip', api_key=new_api_key.api_key),
 
                             status=200)
 

	
 
    def test_access_page_via_expired_api_key(self):
 
        whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
 
        with mock.patch('kallithea.CONFIG', whitelist):
 
            self.assertEqual(['ChangesetController:changeset_raw'],
 
                             whitelist['api_access_controllers_whitelist'])
 

	
 
            new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
 
            Session().commit()
 
            #patch the API key and make it expired
 
            new_api_key.expires = 0
 
            Session().add(new_api_key)
 
            Session().commit()
 
            with fixture.anon_access(False):
 
                self.app.get(url(controller='changeset',
 
                                 action='changeset_raw',
 
                                 repo_name=HG_REPO, revision='tip',
 
                                 api_key=new_api_key.api_key),
 
                             status=302)
0 comments (0 inline, 0 general)