tabpy.tabpy_server.handlers.base_handler.BaseHandler.options() - Code Metrics - Inspection of "Hotfix: model redeployment (#462)" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f89136...5df0e0 )

by Oleksandr

created 2020-11-10 18:28 UTC

BaseHandler.options() A

↳ Parent: tabpy.tabpy_server.handlers.base_handler

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.2963

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	4
ccs	1
cts	3
cp	0.3333
rs	10
c	0
b	0
f	0
cc	1
nop	1
crap	1.2963

import base64
import binascii
import concurrent
import json
import logging
import tornado.web
from tabpy.tabpy_server.app.app_parameters import SettingsParameters
from tabpy.tabpy_server.handlers.util import hash_password
from tabpy.tabpy_server.handlers.util import AuthErrorStates
import uuid


STAGING_THREAD = concurrent.futures.ThreadPoolExecutor(max_workers=3)


class ContextLoggerWrapper:
    """
    This class appends request context to logged messages.
    """

    @staticmethod
    def _generate_call_id():
        return str(uuid.uuid4())

    def __init__(self, request: tornado.httputil.HTTPServerRequest):
        self.call_id = self._generate_call_id()
        self.set_request(request)

        self.tabpy_username = None
        self.log_request_context = False
        self.request_context_logged = False

    def set_request(self, request: tornado.httputil.HTTPServerRequest):
        """
        Set HTTP(S) request for logger. Headers will be used to
        append request data as client information, Tableau user name, etc.
        """
        self.remote_ip = request.remote_ip
        self.method = request.method
        self.url = request.full_url()

        self.client = request.headers.get("TabPy-Client", None)
        self.tableau_username = request.headers.get("TabPy-User", None)

    def set_tabpy_username(self, tabpy_username: str):
        self.tabpy_username = tabpy_username

    def enable_context_logging(self, enable: bool):
        """
        Enable/disable request context information logging.

        Parameters
        ----------
        enable: bool
            If True request context information will be logged and
            every log entry for a request handler will have call ID
            with it.
        """
        self.log_request_context = enable

    def _log_context_info(self):
        if not self.log_request_context:
            return

        context = f"Call ID: {self.call_id}"

        if self.remote_ip is not None:
            context += f", Caller: {self.remote_ip}"

        if self.method is not None:
            context += f", Method: {self.method}"

        if self.url is not None:
            context += f", URL: {self.url}"

        if self.client is not None:
            context += f", Client: {self.client}"

        if self.tableau_username is not None:
            context += f", Tableau user: {self.tableau_username}"

        if self.tabpy_username is not None:
            context += f", TabPy user: {self.tabpy_username}"

        logging.getLogger(__name__).log(logging.INFO, context)
        self.request_context_logged = True

    def log(self, level: int, msg: str):
        """
        Log message with or without call ID. If call context is logged and
        call ID added to any log entry is specified by if context logging
        is enabled (see CallContext.enable_context_logging for more details).

        Parameters
        ----------
        level: int
            Log level: logging.CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET.

        msg: str
            Message format string.

        args
            Same as args in Logger.debug().

        kwargs
            Same as kwargs in Logger.debug().
        """
        extended_msg = msg
        if self.log_request_context:
            if not self.request_context_logged:
                self._log_context_info()

            extended_msg += f", <<call ID: {self.call_id}>>"

        logging.getLogger(__name__).log(level, extended_msg)


class BaseHandler(tornado.web.RequestHandler):
    def initialize(self, app):
        self.tabpy_state = app.tabpy_state
        # set content type to application/json
        self.set_header("Content-Type", "application/json")
        self.protocol = self.settings[SettingsParameters.TransferProtocol]
        self.port = self.settings[SettingsParameters.Port]
        self.python_service = app.python_service
        self.credentials = app.credentials
        self.username = None
        self.password = None
        self.eval_timeout = self.settings[SettingsParameters.EvaluateTimeout]

        self.logger = ContextLoggerWrapper(self.request)
        self.logger.enable_context_logging(
            app.settings[SettingsParameters.LogRequestContext]
        )
        self.logger.log(logging.DEBUG, "Checking if need to handle authentication")
        self.auth_error = self.handle_authentication("v1")

    def error_out(self, code, log_message, info=None):
        self.set_status(code)
        self.write(json.dumps({"message": log_message, "info": info or {}}))

        self.logger.log(
            logging.ERROR,
            'Responding with status={}, message="{}", info="{}"'.format(
                code, log_message, info
            ),
        )

    def options(self):
        # add CORS headers if TabPy has a cors_origin specified
        self._add_CORS_header()
        self.write({})

    def _add_CORS_header(self):
        """
        Add CORS header if the TabPy has attribute _cors_origin
        and _cors_origin is not an empty string.
        """
        origin = self.tabpy_state.get_access_control_allow_origin()
        if len(origin) > 0:
            self.set_header("Access-Control-Allow-Origin", origin)
            self.logger.log(logging.DEBUG, f"Access-Control-Allow-Origin:{origin}")

        headers = self.tabpy_state.get_access_control_allow_headers()
        if len(headers) > 0:
            self.set_header("Access-Control-Allow-Headers", headers)
            self.logger.log(logging.DEBUG, f"Access-Control-Allow-Headers:{headers}")

        methods = self.tabpy_state.get_access_control_allow_methods()
        if len(methods) > 0:
            self.set_header("Access-Control-Allow-Methods", methods)
            self.logger.log(logging.DEBUG, f"Access-Control-Allow-Methods:{methods}")

    def _get_auth_method(self, api_version) -> (bool, str):
        """
        Finds authentication method if provided.

        Parameters
        ----------
        api_version : str
            API version for authentication.

        Returns
        -------
        bool
            True if known authentication method is found.
            False otherwise.

        str
            Name of authentication method used by client.
            If empty no authentication required.

        (True, '') as result of this function means authentication
        is not needed.
        """
        if api_version not in self.settings[SettingsParameters.ApiVersions]:
            self.logger.log(logging.CRITICAL, f'Unknown API version "{api_version}"')
            return False, ""

        version_settings = self.settings[SettingsParameters.ApiVersions][api_version]
        if "features" not in version_settings:
            self.logger.log(
                logging.INFO, f'No features configured for API "{api_version}"'
            )
            return True, ""

        features = version_settings["features"]
        if (
            "authentication" not in features
            or not features["authentication"]["required"]
        ):
            self.logger.log(
                logging.INFO,
                "Authentication is not a required feature for API " f'"{api_version}"',
            )
            return True, ""

        auth_feature = features["authentication"]
        if "methods" not in auth_feature:
            self.logger.log(
                logging.INFO,
                "Authentication method is not configured for API " f'"{api_version}"',
            )

        methods = auth_feature["methods"]
        if "basic-auth" in auth_feature["methods"]:
            return True, "basic-auth"
        # Add new methods here...

        # No known methods were found
        self.logger.log(
            logging.CRITICAL,
            f'Unknown authentication method(s) "{methods}" are configured '
            f'for API "{api_version}"',
        )
        return False, ""

    def _get_basic_auth_credentials(self) -> bool:
        """
        Find credentials for basic access authentication method. Credentials if
        found stored in Credentials.username and Credentials.password.

        Returns
        -------
        bool
            True if valid credentials were found.
            False otherwise.
        """
        self.logger.log(
            logging.DEBUG, "Checking request headers for authentication data"
        )
        if "Authorization" not in self.request.headers:
            self.logger.log(logging.INFO, "Authorization header not found")
            return False

        auth_header = self.request.headers["Authorization"]
        auth_header_list = auth_header.split(" ")
        if len(auth_header_list) != 2 or auth_header_list[0] != "Basic":
            self.logger.log(
                logging.ERROR, f'Unknown authentication method "{auth_header}"'
            )
            return False

        try:
            cred = base64.b64decode(auth_header_list[1]).decode("utf-8")
        except (binascii.Error, UnicodeDecodeError) as ex:
            self.logger.log(logging.CRITICAL, f"Cannot decode credentials: {str(ex)}")
            return False

        login_pwd = cred.split(":")
        if len(login_pwd) != 2:
            self.logger.log(logging.ERROR, "Invalid string in encoded credentials")
            return False

        self.username = login_pwd[0]
        self.logger.set_tabpy_username(self.username)
        self.password = login_pwd[1]
        return True

    def _get_credentials(self, method) -> bool:
        """
        Find credentials for specified authentication method. Credentials if
        found stored in self.username and self.password.

        Parameters
        ----------
        method: str
            Authentication method name.

        Returns
        -------
        bool
            True if valid credentials were found.
            False otherwise.
        """
        if method == "basic-auth":
            return self._get_basic_auth_credentials()
        # Add new methods here...

        # No known methods were found
        self.logger.log(
            logging.CRITICAL,
            f'Unknown authentication method(s) "{method}" are configured ',
        )
        return False

    def _validate_basic_auth_credentials(self) -> bool:
        """
        Validates username:pwd if they are the same as
        stored credentials.

        Returns
        -------
        bool
            True if credentials has key login and
            credentials[login] equal SHA3(pwd), False
            otherwise.
        """
        login = self.username.lower()
        self.logger.log(
            logging.DEBUG, f'Validating credentials for user name "{login}"'
        )
        if login not in self.credentials:
            self.logger.log(logging.ERROR, f'User name "{self.username}" not found')
            return False

        hashed_pwd = hash_password(login, self.password)
        if self.credentials[login].lower() != hashed_pwd.lower():
            self.logger.log(
                logging.ERROR, f'Wrong password for user name "{self.username}"'
            )
            return False

        return True

    def _validate_credentials(self, method) -> bool:
        """
        Validates credentials according to specified methods if they
        are what expected.

        Parameters
        ----------
        method: str
            Authentication method name.

        Returns
        -------
        bool
            True if credentials are valid.
            False otherwise.
        """
        if method == "basic-auth":
            return self._validate_basic_auth_credentials()
        # Add new methods here...

        # No known methods were found
        self.logger.log(
            logging.CRITICAL,
            f'Unknown authentication method(s) "{method}" are configured ',
        )
        return False

    def handle_authentication(self, api_version):
        """
        If authentication feature is configured checks provided
        credentials.

        Parameters
        ----------
        api_version : str
            API version for authentication.

        Returns
        -------
        String
            None if authentication is not required and username and password are None.
            None if authentication is required and valid credentials provided.
            NotAuthorized if authenication is required and credentials are incorrect.
            NotRequired if authentication is not required but credentials are provided.
        """
        self.logger.log(logging.DEBUG, "Handling authentication")
        found, method = self._get_auth_method(api_version)
        if not found:
            return AuthErrorStates.NotAuthorized

        if method == "":
            if not self._get_basic_auth_credentials():
                self.logger.log(logging.DEBUG,
                                "authentication not required, username and password are none")
                return AuthErrorStates.NONE
            else:
                self.logger.log(logging.DEBUG,
                                "authentication not required, username and password are not none")
                return AuthErrorStates.NotRequired

        if not self._get_credentials(method):
            return AuthErrorStates.NotAuthorized

        if not self._validate_credentials(method):
            return AuthErrorStates.NotAuthorized

        return AuthErrorStates.NONE

    def should_fail_with_auth_error(self):
        """
        Checks if authentication is required:
        - if it is not returns false, None
        - if it is required validates provided credentials

        Returns
        -------
        bool
            False if authentication is not required and username
            and password is None or isrequired and validation
            for credentials passes.
            True if validation for credentials failed or
            if authentication is not required and username and password
            fields are not empty.
        """
        return self.auth_error

    def fail_with_auth_error(self):
        """
        Prepares server 401 response and server 400 response depending
        on the value of the self.auth_error flag
        """
        if self.auth_error == AuthErrorStates.NotAuthorized:
            self.logger.log(logging.ERROR, "Failing with 401 for unauthorized request")
            self.set_status(401)
            self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
            self.error_out(
                401,
                info="Unauthorized request.",
                log_message="Invalid credentials provided.",
            )
        else:
            self.logger.log(logging.ERROR, "Failing with 400 for Bad Request")
            self.set_status(400)
            self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
            self.error_out(
                400,
                info="Bad request.",
                log_message="Username or Password provided when authentication not available",
            )


1	1	import base64
2	1	import binascii
3	1	import concurrent
4	1	import json
5	1	import logging
6	1	import tornado.web
7	1	from tabpy.tabpy_server.app.app_parameters import SettingsParameters
8	1	from tabpy.tabpy_server.handlers.util import hash_password
9	1	from tabpy.tabpy_server.handlers.util import AuthErrorStates
10	1	import uuid
11
12
13	1	STAGING_THREAD = concurrent.futures.ThreadPoolExecutor(max_workers=3)
14
15
16	1	class ContextLoggerWrapper:
17		"""
18		This class appends request context to logged messages.
19		"""
20
21	1	@staticmethod
22		def _generate_call_id():
23	1	return str(uuid.uuid4())
24
25	1	def __init__(self, request: tornado.httputil.HTTPServerRequest):
26	1	self.call_id = self._generate_call_id()
27	1	self.set_request(request)
28
29	1	self.tabpy_username = None
30	1	self.log_request_context = False
31	1	self.request_context_logged = False
32
33	1	def set_request(self, request: tornado.httputil.HTTPServerRequest):
34		"""
35		Set HTTP(S) request for logger. Headers will be used to
36		append request data as client information, Tableau user name, etc.
37		"""
38	1	self.remote_ip = request.remote_ip
39	1	self.method = request.method
40	1	self.url = request.full_url()
41
42	1	self.client = request.headers.get("TabPy-Client", None)
43	1	self.tableau_username = request.headers.get("TabPy-User", None)
44
45	1	def set_tabpy_username(self, tabpy_username: str):
46	1	self.tabpy_username = tabpy_username
47
48	1	def enable_context_logging(self, enable: bool):
49		"""
50		Enable/disable request context information logging.
51
52		Parameters
53		----------
54		enable: bool
55		If True request context information will be logged and
56		every log entry for a request handler will have call ID
57		with it.
58		"""
59	1	self.log_request_context = enable
60
61	1	def _log_context_info(self):
62		if not self.log_request_context:
63		return
64
65		context = f"Call ID: {self.call_id}"
66
67		if self.remote_ip is not None:
68		context += f", Caller: {self.remote_ip}"
69
70		if self.method is not None:
71		context += f", Method: {self.method}"
72
73		if self.url is not None:
74		context += f", URL: {self.url}"
75
76		if self.client is not None:
77		context += f", Client: {self.client}"
78
79		if self.tableau_username is not None:
80		context += f", Tableau user: {self.tableau_username}"
81
82		if self.tabpy_username is not None:
83		context += f", TabPy user: {self.tabpy_username}"
84
85		logging.getLogger(__name__).log(logging.INFO, context)
86		self.request_context_logged = True
87
88	1	def log(self, level: int, msg: str):
89		"""
90		Log message with or without call ID. If call context is logged and
91		call ID added to any log entry is specified by if context logging
92		is enabled (see CallContext.enable_context_logging for more details).
93
94		Parameters
95		----------
96		level: int
97		Log level: logging.CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET.
98
99		msg: str
100		Message format string.
101
102		args
103		Same as args in Logger.debug().
104
105		kwargs
106		Same as kwargs in Logger.debug().
107		"""
108	1	extended_msg = msg
109	1	if self.log_request_context:
110		if not self.request_context_logged:
111		self._log_context_info()
112
113		extended_msg += f", <<call ID: {self.call_id}>>"
114
115	1	logging.getLogger(__name__).log(level, extended_msg)
116
117
118	1	class BaseHandler(tornado.web.RequestHandler):
119	1	def initialize(self, app):
120	1	self.tabpy_state = app.tabpy_state
121		# set content type to application/json
122	1	self.set_header("Content-Type", "application/json")
123	1	self.protocol = self.settings[SettingsParameters.TransferProtocol]
124	1	self.port = self.settings[SettingsParameters.Port]
125	1	self.python_service = app.python_service
126	1	self.credentials = app.credentials
127	1	self.username = None
128	1	self.password = None
129	1	self.eval_timeout = self.settings[SettingsParameters.EvaluateTimeout]
130
131	1	self.logger = ContextLoggerWrapper(self.request)
132	1	self.logger.enable_context_logging(
133		app.settings[SettingsParameters.LogRequestContext]
134		)
135	1	self.logger.log(logging.DEBUG, "Checking if need to handle authentication")
136	1	self.auth_error = self.handle_authentication("v1")
137
138	1	def error_out(self, code, log_message, info=None):
139	1	self.set_status(code)
140	1	self.write(json.dumps({"message": log_message, "info": info or {}}))
141
142	1	self.logger.log(
143		logging.ERROR,
144		'Responding with status={}, message="{}", info="{}"'.format(
145		code, log_message, info
146		),
147		)
148
149	1	def options(self):
150		# add CORS headers if TabPy has a cors_origin specified
151		self._add_CORS_header()
152		self.write({})
153
154	1	def _add_CORS_header(self):
155		"""
156		Add CORS header if the TabPy has attribute _cors_origin
157		and _cors_origin is not an empty string.
158		"""
159	1	origin = self.tabpy_state.get_access_control_allow_origin()
160	1	if len(origin) > 0:
161		self.set_header("Access-Control-Allow-Origin", origin)
162		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Origin:{origin}")
163
164	1	headers = self.tabpy_state.get_access_control_allow_headers()
165	1	if len(headers) > 0:
166		self.set_header("Access-Control-Allow-Headers", headers)
167		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Headers:{headers}")
168
169	1	methods = self.tabpy_state.get_access_control_allow_methods()
170	1	if len(methods) > 0:
171		self.set_header("Access-Control-Allow-Methods", methods)
172		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Methods:{methods}")
173
174	1	def _get_auth_method(self, api_version) -> (bool, str):
175		"""
176		Finds authentication method if provided.
177
178		Parameters
179		----------
180		api_version : str
181		API version for authentication.
182
183		Returns
184		-------
185		bool
186		True if known authentication method is found.
187		False otherwise.
188
189		str
190		Name of authentication method used by client.
191		If empty no authentication required.
192
193		(True, '') as result of this function means authentication
194		is not needed.
195		"""
196	1	if api_version not in self.settings[SettingsParameters.ApiVersions]:
197		self.logger.log(logging.CRITICAL, f'Unknown API version "{api_version}"')
198		return False, ""
199
200	1	version_settings = self.settings[SettingsParameters.ApiVersions][api_version]
201	1	if "features" not in version_settings:
202		self.logger.log(
203		logging.INFO, f'No features configured for API "{api_version}"'
204		)
205		return True, ""
206
207	1	features = version_settings["features"]
208	1	if (
209		"authentication" not in features
210		or not features["authentication"]["required"]
211		):
212	1	self.logger.log(
213		logging.INFO,
214		"Authentication is not a required feature for API " f'"{api_version}"',
215		)
216	1	return True, ""
217
218	1	auth_feature = features["authentication"]
219	1	if "methods" not in auth_feature:
220		self.logger.log(
221		logging.INFO,
222		"Authentication method is not configured for API " f'"{api_version}"',
223		)
224
225	1	methods = auth_feature["methods"]
226	1	if "basic-auth" in auth_feature["methods"]:
227	1	return True, "basic-auth"
228		# Add new methods here...
229
230		# No known methods were found
231		self.logger.log(
232		logging.CRITICAL,
233		f'Unknown authentication method(s) "{methods}" are configured '
234		f'for API "{api_version}"',
235		)
236		return False, ""
237
238	1	def _get_basic_auth_credentials(self) -> bool:
239		"""
240		Find credentials for basic access authentication method. Credentials if
241		found stored in Credentials.username and Credentials.password.
242
243		Returns
244		-------
245		bool
246		True if valid credentials were found.
247		False otherwise.
248		"""
249	1	self.logger.log(
250		logging.DEBUG, "Checking request headers for authentication data"
251		)
252	1	if "Authorization" not in self.request.headers:
253	1	self.logger.log(logging.INFO, "Authorization header not found")
254	1	return False
255
256	1	auth_header = self.request.headers["Authorization"]
257	1	auth_header_list = auth_header.split(" ")
258	1	if len(auth_header_list) != 2 or auth_header_list[0] != "Basic":
259		self.logger.log(
260		logging.ERROR, f'Unknown authentication method "{auth_header}"'
261		)
262		return False
263
264	1	try:
265	1	cred = base64.b64decode(auth_header_list[1]).decode("utf-8")
266		except (binascii.Error, UnicodeDecodeError) as ex:
267		self.logger.log(logging.CRITICAL, f"Cannot decode credentials: {str(ex)}")
268		return False
269
270	1	login_pwd = cred.split(":")
271	1	if len(login_pwd) != 2:
272		self.logger.log(logging.ERROR, "Invalid string in encoded credentials")
273		return False
274
275	1	self.username = login_pwd[0]
276	1	self.logger.set_tabpy_username(self.username)
277	1	self.password = login_pwd[1]
278	1	return True
279
280	1	def _get_credentials(self, method) -> bool:
281		"""
282		Find credentials for specified authentication method. Credentials if
283		found stored in self.username and self.password.
284
285		Parameters
286		----------
287		method: str
288		Authentication method name.
289
290		Returns
291		-------
292		bool
293		True if valid credentials were found.
294		False otherwise.
295		"""
296	1	if method == "basic-auth":
297	1	return self._get_basic_auth_credentials()
298		# Add new methods here...
299
300		# No known methods were found
301		self.logger.log(
302		logging.CRITICAL,
303		f'Unknown authentication method(s) "{method}" are configured ',
304		)
305		return False
306
307	1	def _validate_basic_auth_credentials(self) -> bool:
308		"""
309		Validates username:pwd if they are the same as
310		stored credentials.
311
312		Returns
313		-------
314		bool
315		True if credentials has key login and
316		credentials[login] equal SHA3(pwd), False
317		otherwise.
318		"""
319	1	login = self.username.lower()
320	1	self.logger.log(
321		logging.DEBUG, f'Validating credentials for user name "{login}"'
322		)
323	1	if login not in self.credentials:
324	1	self.logger.log(logging.ERROR, f'User name "{self.username}" not found')
325	1	return False
326
327	1	hashed_pwd = hash_password(login, self.password)
328	1	if self.credentials[login].lower() != hashed_pwd.lower():
329		self.logger.log(
330		logging.ERROR, f'Wrong password for user name "{self.username}"'
331		)
332		return False
333
334	1	return True
335
336	1	def _validate_credentials(self, method) -> bool:
337		"""
338		Validates credentials according to specified methods if they
339		are what expected.
340
341		Parameters
342		----------
343		method: str
344		Authentication method name.
345
346		Returns
347		-------
348		bool
349		True if credentials are valid.
350		False otherwise.
351		"""
352	1	if method == "basic-auth":
353	1	return self._validate_basic_auth_credentials()
354		# Add new methods here...
355
356		# No known methods were found
357		self.logger.log(
358		logging.CRITICAL,
359		f'Unknown authentication method(s) "{method}" are configured ',
360		)
361		return False
362
363	1	def handle_authentication(self, api_version):
364		"""
365		If authentication feature is configured checks provided
366		credentials.
367
368		Parameters
369		----------
370		api_version : str
371		API version for authentication.
372
373		Returns
374		-------
375		String
376		None if authentication is not required and username and password are None.
377		None if authentication is required and valid credentials provided.
378		NotAuthorized if authenication is required and credentials are incorrect.
379		NotRequired if authentication is not required but credentials are provided.
380		"""
381	1	self.logger.log(logging.DEBUG, "Handling authentication")
382	1	found, method = self._get_auth_method(api_version)
383	1	if not found:
384		return AuthErrorStates.NotAuthorized
385
386	1	if method == "":
387	1	if not self._get_basic_auth_credentials():
388	1	self.logger.log(logging.DEBUG,
389		"authentication not required, username and password are none")
390	1	return AuthErrorStates.NONE
391		else:
392	1	self.logger.log(logging.DEBUG,
393		"authentication not required, username and password are not none")
394	1	return AuthErrorStates.NotRequired
395
396	1	if not self._get_credentials(method):
397	1	return AuthErrorStates.NotAuthorized
398
399	1	if not self._validate_credentials(method):
400	1	return AuthErrorStates.NotAuthorized
401
402	1	return AuthErrorStates.NONE
403
404	1	def should_fail_with_auth_error(self):
405		"""
406		Checks if authentication is required:
407		- if it is not returns false, None
408		- if it is required validates provided credentials
409
410		Returns
411		-------
412		bool
413		False if authentication is not required and username
414		and password is None or isrequired and validation
415		for credentials passes.
416		True if validation for credentials failed or
417		if authentication is not required and username and password
418		fields are not empty.
419		"""
420	1	return self.auth_error
421
422	1	def fail_with_auth_error(self):
423		"""
424		Prepares server 401 response and server 400 response depending
425		on the value of the self.auth_error flag
426		"""
427	1	if self.auth_error == AuthErrorStates.NotAuthorized:
428	1	self.logger.log(logging.ERROR, "Failing with 401 for unauthorized request")
429	1	self.set_status(401)
430	1	self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
431	1	self.error_out(
432		401,
433		info="Unauthorized request.",
434		log_message="Invalid credentials provided.",
435		)
436		else:
437	1	self.logger.log(logging.ERROR, "Failing with 400 for Bad Request")
438	1	self.set_status(400)
439	1	self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
440	1	self.error_out(
441		400,
442		info="Bad request.",
443		log_message="Username or Password provided when authentication not available",
444		)
445

tableau / TabPy

Push — master ( f89136...5df0e0 )

BaseHandler.options() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like