code/smart_httpd.py at master · qingfeng/code · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#!/usr/bin/python
# coding:utf8

import os.path

from wsgiauth.basic import BasicAuth

from code.config import DEVELOP_MODE
from code.libs.git_http_backend import assemble_WSGI_git_app
from code.libs.permdir import get_repo_root
from code.models.user import User
from code.models.project import Project
from code.models.organization import Organization


DOUBAN_REALM = "douban wsgi basic auth"


def get_proj(path):
    '''
    >>> get_proj('/a.git/info/refs')
    'a'
    >>> get_proj('/testuser/a.git/info/refs')
    'testuser/a'
    '''
    path_split = path.split("/")
    git_name = path_split[1]
    if not git_name.endswith('.git'):
        git_name = "/".join(path_split[1:3])
    assert "." in git_name, "Need a string looking like project.git, got '%s'" % git_name
    proj_name, _ = os.path.splitext(git_name)
    return proj_name


def get_git_path_info(path):
    path_split = path.split("/")
    git_name = path_split[1]
    # raw path: project_id.git
    if git_name.endswith('.git'):
        project = Project.get_by_name(git_name[:-4])
        if project:
            path_split[1] = "%s.git" % project.id
            return '/'.join(path_split)
    else:
        owner_name, git_name = path_split[1:3]
        # user project: user/project.git
        user = User.get_by_name(owner_name)
        if user:
            project = Project.get_by_name_and_owner(git_name[:-4], user.id)
            if project:
                path_split[1] = ""
                path_split[2] = "%s.git" % project.id
                return '/'.join(path_split[1:])
            return
        # org project: org/project.git
        org = Organization.get_by_name(owner_name)
        if org:
            project = Project.get_by_name_and_owner(git_name[:-4], user.id)
            if project:
                path_split[1] = ""
                path_split[2] = "%s.git" % project.id
                return '/'.join(path_split[1:])


def authfunc(environ, username, passwd):
    # no auth
    return True

    if DEVELOP_MODE or (environ['REMOTE_ADDR'] == '127.0.0.1'
                        and environ['HTTP_HOST'] == 'localhost:8080'):
        return True
    if not passwd:
        return
    if username == 'code' and passwd == 'code':
        return True
    # FIXME: login
    is_good = True
    if not is_good:
        return
    is_push = 'service=git-receive-pack' in environ['QUERY_STRING'] or '/git-receive-pack' in environ['PATH_INFO']
    if is_push:
        pass
        # FIXME: push permission
    return True


class HTTPAuth(object):
    def __init__(self, application, realm, authfunc, scheme, **kw):
        self.application = application
        self.authenticate = scheme(realm, authfunc, **kw)
        self.scheme = scheme.authtype

    def __call__(self, environ, start_response):
        path_info = environ['PATH_INFO']
        environ['GIT_PATH_INFO'] = get_git_path_info(path_info)
        if environ.get('REMOTE_USER') is None:
            result = self.authenticate(environ)
            if not isinstance(result, str):
                # Request credentials if authentication fails
                return result(environ, start_response)
            environ['REMOTE_USER'] = result
            environ['AUTH_TYPE'] = self.scheme
        return self.application(environ, start_response)


class WebRedirect(object):
    def __init__(self, application, **kw):
        self.application = application

    def __call__(self, environ, start_response):
        http_accept = environ.get('HTTP_ACCEPT')
        path_info = environ.get('PATH_INFO')
        proj_name = get_proj(path_info)
        if http_accept and 'text/html' in http_accept:
            start_response('301 Redirect',
                           [('Location', '/%s/' % proj_name), ])
            return []
        return self.application(environ, start_response)


app = assemble_WSGI_git_app(content_path=get_repo_root())
app = WebRedirect(app)
app = HTTPAuth(app, DOUBAN_REALM, authfunc, BasicAuth)