"""The base Controller API
Provides the BaseController class for subclassing.
"""
import logging
import time
import traceback
from paste.auth.basic import AuthBasicAuthenticator
from paste.httpexceptions import HTTPUnauthorized, HTTPForbidden
from webob.exc import HTTPClientError
from paste.httpheaders import WWW_AUTHENTICATE
from paste.httpheaders import WWW_AUTHENTICATE, AUTHORIZATION
from pylons import config, tmpl_context as c, request, session, url
from pylons.controllers import WSGIController
from pylons.controllers.util import redirect
from pylons.templating import render_mako as render
from rhodecode import __version__, BACKENDS
from rhodecode.lib.utils2 import str2bool, safe_unicode, AttributeDict,\
safe_str
from rhodecode.lib.auth import AuthUser, get_container_username, authfunc,\
HasPermissionAnyMiddleware, CookieStoreWrapper
from rhodecode.lib.utils import get_repo_slug, invalidate_cache
from rhodecode.model import meta
from rhodecode.model.db import Repository, RhodeCodeUi, User
from rhodecode.model.notification import NotificationModel
from rhodecode.model.scm import ScmModel
from rhodecode.model.meta import Session
log = logging.getLogger(__name__)
def _get_ip_addr(environ):
proxy_key = 'HTTP_X_REAL_IP'
proxy_key2 = 'HTTP_X_FORWARDED_FOR'
def_key = 'REMOTE_ADDR'
ip = environ.get(proxy_key2)
if ip:
return ip
ip = environ.get(proxy_key)
ip = environ.get(def_key, '0.0.0.0')
def _get_access_path(environ):
path = environ.get('PATH_INFO')
org_req = environ.get('pylons.original_request')
if org_req:
path = org_req.environ.get('PATH_INFO')
return path
class BasicAuth(AuthBasicAuthenticator):
def __init__(self, realm, authfunc, auth_http_code=None):
self.realm = realm
self.authfunc = authfunc
self._rc_auth_http_code = auth_http_code
def build_authentication(self):
head = WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
if self._rc_auth_http_code and self._rc_auth_http_code == '403':
# return 403 if alternative http return code is specified in
# RhodeCode config
return HTTPForbidden(headers=head)
return HTTPUnauthorized(headers=head)
def authenticate(self, environ):
authorization = AUTHORIZATION(environ)
if not authorization:
return self.build_authentication()
(authmeth, auth) = authorization.split(' ', 1)
if 'basic' != authmeth.lower():
auth = auth.strip().decode('base64')
_parts = auth.split(':', 1)
if len(_parts) == 2:
username, password = _parts
if self.authfunc(environ, username, password):
return username
__call__ = authenticate
class BaseVCSController(object):
def __init__(self, application, config):
self.application = application
self.config = config
# base path of repo locations
self.basepath = self.config['base_path']
#authenticate this mercurial request using authfunc
self.authenticate = BasicAuth('', authfunc,
config.get('auth_ret_code'))
self.ipaddr = '0.0.0.0'
def _handle_request(self, environ, start_response):
raise NotImplementedError()
def _get_by_id(self, repo_name):
Get's a special pattern _<ID> from clone url and tries to replace it
with a repository_name for support of _<ID> non changable urls
:param repo_name:
try:
data = repo_name.split('/')
if len(data) >= 2:
by_id = data[1].split('_')
if len(by_id) == 2 and by_id[1].isdigit():
_repo_name = Repository.get(by_id[1]).repo_name
data[1] = _repo_name
except:
log.debug('Failed to extract repo_name from id %s' % (
traceback.format_exc()
)
return '/'.join(data)
def _invalidate_cache(self, repo_name):
Set's cache for this repository for invalidation on next access
:param repo_name: full repo name, also a cache key
invalidate_cache('get_repo_cached_%s' % repo_name)
def _check_permission(self, action, user, repo_name):
Status change: