NeedAuthController.authDomainUser() - Code Metrics - Inspection of "need_auth will work on /dav interface as well. fix..." - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f81e82...0af933 )

by Roy

created 2017-01-19 15:11 UTC

NeedAuthController.authDomainUser() A

↳ Parent: NeedAuthController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	3
rs	10

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
# Author: Binux<[email protected]>
#         http://binux.me
# Created on 2015-6-3 11:29


import os
import time
import base64
import six
from six import BytesIO
from wsgidav.wsgidav_app import DEFAULT_CONFIG, WsgiDAVApp
from wsgidav.dav_provider import DAVProvider, DAVCollection, DAVNonCollection
from wsgidav.dav_error import DAVError, HTTP_FORBIDDEN
from pyspider.libs.utils import utf8, text
from .app import app


def check_user(environ):
    authheader = environ.get("HTTP_AUTHORIZATION")
    if not authheader:
        return False
    authheader = authheader[len("Basic "):]
    try:
        username, password = text(base64.b64decode(authheader)).split(':', 1)
    except Exception as e:
        app.logger.error('wrong api key: %r, %r', authheader, e)
        return False

    if username == app.config['webui_username'] \
            and password == app.config['webui_password']:
        return True
    else:
        return False


class ContentIO(BytesIO):
    def close(self):
        self.content = self.getvalue()
        BytesIO.close(self)


class ScriptResource(DAVNonCollection):
    def __init__(self, path, environ, app, project=None):
        super(ScriptResource, self).__init__(path, environ)

        self.app = app
        self.new_project = False
        self._project = project
        self.project_name = text(self.name)
        self.writebuffer = None
        if self.project_name.endswith('.py'):
            self.project_name = self.project_name[:-len('.py')]

    @property
    def project(self):
        if self._project:
            return self._project
        projectdb = self.app.config['projectdb']
        if projectdb:
            self._project = projectdb.get(self.project_name)
        if not self._project:
            if projectdb.verify_project_name(self.project_name) and text(self.name).endswith('.py'):
                self.new_project = True
                self._project = {
                    'name': self.project_name,
                    'script': '',
                    'status': 'TODO',
                    'rate': self.app.config.get('max_rate', 1),
                    'burst': self.app.config.get('max_burst', 3),
                    'updatetime': time.time(),
                }
            else:
                raise DAVError(HTTP_FORBIDDEN)
        return self._project

    @property
    def readonly(self):
        projectdb = self.app.config['projectdb']
        if not projectdb:
            return True
        if 'lock' in projectdb.split_group(self.project.get('group')) \
                and self.app.config.get('webui_username') \
                and self.app.config.get('webui_password'):
            return not check_user(self.environ)
        return False

    def getContentLength(self):
        return len(utf8(self.project['script']))

    def getContentType(self):
        return 'text/plain'

    def getLastModified(self):
        return self.project['updatetime']

    def getContent(self):
        return BytesIO(utf8(self.project['script']))

    def beginWrite(self, contentType=None):
        if self.readonly:
            self.app.logger.error('webdav.beginWrite readonly')
            return super(ScriptResource, self).beginWrite(contentType)
        self.writebuffer = ContentIO()
        return self.writebuffer

    def endWrite(self, withErrors):
        if withErrors:
            self.app.logger.error('webdav.endWrite error: %r', withErrors)
            return super(ScriptResource, self).endWrite(withErrors)
        if not self.writebuffer:
            return
        projectdb = self.app.config['projectdb']
        if not projectdb:
            return

        info = {
            'script': text(getattr(self.writebuffer, 'content', ''))
        }
        if self.project.get('status') in ('DEBUG', 'RUNNING'):
            info['status'] = 'CHECKING'

        if self.new_project:
            self.project.update(info)
            self.new_project = False
            return projectdb.insert(self.project_name, self.project)
        else:
            return projectdb.update(self.project_name, info)


class RootCollection(DAVCollection):
    def __init__(self, path, environ, app):
        super(RootCollection, self).__init__(path, environ)
        self.app = app
        self.projectdb = self.app.config['projectdb']

    def getMemberList(self):
        members = []
        for project in self.projectdb.get_all():
            project_name = project['name']
            if not project_name.endswith('.py'):
                project_name += '.py'
            native_path = os.path.join(self.path, project_name)
            native_path = text(native_path) if six.PY3 else utf8(native_path)
            members.append(ScriptResource(
                native_path,
                self.environ,
                self.app,
                project
            ))
        return members

    def getMemberNames(self):
        members = []
        for project in self.projectdb.get_all(fields=['name', ]):
            project_name = project['name']
            if not project_name.endswith('.py'):
                project_name += '.py'
            members.append(utf8(project_name))
        return members


class ScriptProvider(DAVProvider):
    def __init__(self, app):
        super(ScriptProvider, self).__init__()
        self.app = app

    def __repr__(self):
        return "pyspiderScriptProvider"

    def getResourceInst(self, path, environ):
        path = os.path.normpath(path).replace('\\', '/')
        if path in ('/', '.', ''):
            path = '/'
            return RootCollection(path, environ, self.app)
        else:
            return ScriptResource(path, environ, self.app)


class NeedAuthController(object):
    def __init__(self, app):
        self.app = app

    def getDomainRealm(self, inputRelativeURL, environ):
        return 'need auth'

    def requireAuthentication(self, realmname, environ):
        return self.app.config.get('need_auth', False)

    def isRealmUser(self, realmname, username, environ):
        return username == self.app.config.get('webui_username')

    def getRealmUserPassword(self, realmname, username, environ):
        return self.app.config.get('webui_password')

    def authDomainUser(self, realmname, username, password, environ):
        return username == self.app.config.get('webui_username') \
            and password == self.app.config.get('webui_password')


config = DEFAULT_CONFIG.copy()
config.update({
    'mount_path': '/dav',
    'provider_mapping': {
        '/': ScriptProvider(app)
    },
    'domaincontroller': NeedAuthController(app),
    'verbose': 1 if app.debug else 0,
    'dir_browser': {'davmount': False,
                    'enable': True,
                    'msmount': False,
                    'response_trailer': ''},
})
dav_app = WsgiDAVApp(config)


1			#!/usr/bin/env python
2			# -- encoding: utf-8 --
3			# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
4			# Author: Binux<[email protected]>
5			# http://binux.me
6			# Created on 2015-6-3 11:29
7
8
9			import os
10			import time
11			import base64
12			import six
13			from six import BytesIO
14			from wsgidav.wsgidav_app import DEFAULT_CONFIG, WsgiDAVApp
15			from wsgidav.dav_provider import DAVProvider, DAVCollection, DAVNonCollection
16			from wsgidav.dav_error import DAVError, HTTP_FORBIDDEN
17			from pyspider.libs.utils import utf8, text
18			from .app import app
19
20
21			def check_user(environ):
22			authheader = environ.get("HTTP_AUTHORIZATION")
23			if not authheader:
24			return False
25			authheader = authheader[len("Basic "):]
26			try:
27			username, password = text(base64.b64decode(authheader)).split(':', 1)
28			except Exception as e:
29			app.logger.error('wrong api key: %r, %r', authheader, e)
30			return False
31
32			if username == app.config['webui_username'] \
33			and password == app.config['webui_password']:
34			return True
35			else:
36			return False
37
38
39			class ContentIO(BytesIO):
40			def close(self):
41			self.content = self.getvalue()
42			BytesIO.close(self)
43
44
45			class ScriptResource(DAVNonCollection):
46			def __init__(self, path, environ, app, project=None):
47			super(ScriptResource, self).__init__(path, environ)
48
49			self.app = app
50			self.new_project = False
51			self._project = project
52			self.project_name = text(self.name)
53			self.writebuffer = None
54			if self.project_name.endswith('.py'):
55			self.project_name = self.project_name[:-len('.py')]
56
57			@property
58			def project(self):
59			if self._project:
60			return self._project
61			projectdb = self.app.config['projectdb']
62			if projectdb:
63			self._project = projectdb.get(self.project_name)
64			if not self._project:
65			if projectdb.verify_project_name(self.project_name) and text(self.name).endswith('.py'):
66			self.new_project = True
67			self._project = {
68			'name': self.project_name,
69			'script': '',
70			'status': 'TODO',
71			'rate': self.app.config.get('max_rate', 1),
72			'burst': self.app.config.get('max_burst', 3),
73			'updatetime': time.time(),
74			}
75			else:
76			raise DAVError(HTTP_FORBIDDEN)
77			return self._project
78
79			@property
80			def readonly(self):
81			projectdb = self.app.config['projectdb']
82			if not projectdb:
83			return True
84			if 'lock' in projectdb.split_group(self.project.get('group')) \
85			and self.app.config.get('webui_username') \
86			and self.app.config.get('webui_password'):
87			return not check_user(self.environ)
88			return False
89
90			def getContentLength(self):
91			return len(utf8(self.project['script']))
92
93			def getContentType(self):
94			return 'text/plain'
95
96			def getLastModified(self):
97			return self.project['updatetime']
98
99			def getContent(self):
100			return BytesIO(utf8(self.project['script']))
101
102			def beginWrite(self, contentType=None):
103			if self.readonly:
104			self.app.logger.error('webdav.beginWrite readonly')
105			return super(ScriptResource, self).beginWrite(contentType)
106			self.writebuffer = ContentIO()
107			return self.writebuffer
108
109			def endWrite(self, withErrors):
110			if withErrors:
111			self.app.logger.error('webdav.endWrite error: %r', withErrors)
112			return super(ScriptResource, self).endWrite(withErrors)
113			if not self.writebuffer:
114			return
115			projectdb = self.app.config['projectdb']
116			if not projectdb:
117			return
118
119			info = {
120			'script': text(getattr(self.writebuffer, 'content', ''))
121			}
122			if self.project.get('status') in ('DEBUG', 'RUNNING'):
123			info['status'] = 'CHECKING'
124
125			if self.new_project:
126			self.project.update(info)
127			self.new_project = False
128			return projectdb.insert(self.project_name, self.project)
129			else:
130			return projectdb.update(self.project_name, info)
131
132
133			class RootCollection(DAVCollection):
134			def __init__(self, path, environ, app):
135			super(RootCollection, self).__init__(path, environ)
136			self.app = app
137			self.projectdb = self.app.config['projectdb']
138
139			def getMemberList(self):
140			members = []
141			for project in self.projectdb.get_all():
142			project_name = project['name']
143			if not project_name.endswith('.py'):
144			project_name += '.py'
145			native_path = os.path.join(self.path, project_name)
146			native_path = text(native_path) if six.PY3 else utf8(native_path)
147			members.append(ScriptResource(
148			native_path,
149			self.environ,
150			self.app,
151			project
152			))
153			return members
154
155			def getMemberNames(self):
156			members = []
157			for project in self.projectdb.get_all(fields=['name', ]):
158			project_name = project['name']
159			if not project_name.endswith('.py'):
160			project_name += '.py'
161			members.append(utf8(project_name))
162			return members
163
164
165			class ScriptProvider(DAVProvider):
166			def __init__(self, app):
167			super(ScriptProvider, self).__init__()
168			self.app = app
169
170			def __repr__(self):
171			return "pyspiderScriptProvider"
172
173			def getResourceInst(self, path, environ):
174			path = os.path.normpath(path).replace('\\', '/')
175			if path in ('/', '.', ''):
176			path = '/'
177			return RootCollection(path, environ, self.app)
178			else:
179			return ScriptResource(path, environ, self.app)
180
181
182			class NeedAuthController(object):
183			def __init__(self, app):
184			self.app = app
185
186			def getDomainRealm(self, inputRelativeURL, environ):
187			return 'need auth'
188
189			def requireAuthentication(self, realmname, environ):
190			return self.app.config.get('need_auth', False)
191
192			def isRealmUser(self, realmname, username, environ):
193			return username == self.app.config.get('webui_username')
194
195			def getRealmUserPassword(self, realmname, username, environ):
196			return self.app.config.get('webui_password')
197
198			def authDomainUser(self, realmname, username, password, environ):
199			return username == self.app.config.get('webui_username') \
200			and password == self.app.config.get('webui_password')
201
202
203			config = DEFAULT_CONFIG.copy()
204			config.update({
205			'mount_path': '/dav',
206			'provider_mapping': {
207			'/': ScriptProvider(app)
208			},
209			'domaincontroller': NeedAuthController(app),
210			'verbose': 1 if app.debug else 0,
211			'dir_browser': {'davmount': False,
212			'enable': True,
213			'msmount': False,
214			'response_trailer': ''},
215			})
216			dav_app = WsgiDAVApp(config)
217

binux / pyspider

Push — master ( f81e82...0af933 )

NeedAuthController.authDomainUser() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like