tabpy.tabpy_server.handlers.base_handler.BaseHandler._get_auth_method() - Code Metrics - Inspection of "Running TabPy on AWS" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#566)

unknown

created 2022-12-09 23:18 UTC

tabpy.tabpy_server.handlers.base_handler.BaseHandler._get_auth_method() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	63
Code Lines	29

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	13
CRAP Score	9.1008

Importance

Changes

Metric	Value
eloc	29
dl	0
loc	63
ccs	13
cts	20
cp	0.65
rs	7.784
c	0
b	0
f	0
cc	7
nop	2
crap	9.1008

How to fix Long Method

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.set_header(“Strict-Transport-Security”, “preload; max-age=2592000")
        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 406 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 406 for Not Acceptable")
            self.set_status(406)
            self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
            self.error_out(
                406,
                info="Not Acceptable",
                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.set_header(“Strict-Transport-Security”, “preload; max-age=2592000")
124	1	self.protocol = self.settings[SettingsParameters.TransferProtocol]
125	1	self.port = self.settings[SettingsParameters.Port]
126	1	self.python_service = app.python_service
127	1	self.credentials = app.credentials
128	1	self.username = None
129	1	self.password = None
130		self.eval_timeout = self.settings[SettingsParameters.EvaluateTimeout]
131	1
132	1	self.logger = ContextLoggerWrapper(self.request)
133		self.logger.enable_context_logging(
134		app.settings[SettingsParameters.LogRequestContext]
135	1	)
136	1	self.logger.log(logging.DEBUG, "Checking if need to handle authentication")
137		self.auth_error = self.handle_authentication("v1")
138	1
139	1	def error_out(self, code, log_message, info=None):
140	1	self.set_status(code)
141		self.write(json.dumps({"message": log_message, "info": info or {}}))
142	1
143		self.logger.log(
144		logging.ERROR,
145		'Responding with status={}, message="{}", info="{}"'.format(
146		code, log_message, info
147		),
148		)
149	1
150		def options(self):
151		# add CORS headers if TabPy has a cors_origin specified
152		self._add_CORS_header()
153		self.write({})
154	1
155		def _add_CORS_header(self):
156		"""
157		Add CORS header if the TabPy has attribute _cors_origin
158		and _cors_origin is not an empty string.
159	1	"""
160	1	origin = self.tabpy_state.get_access_control_allow_origin()
161		if len(origin) > 0:
162		self.set_header("Access-Control-Allow-Origin", origin)
163		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Origin:{origin}")
164	1
165	1	headers = self.tabpy_state.get_access_control_allow_headers()
166		if len(headers) > 0:
167		self.set_header("Access-Control-Allow-Headers", headers)
168		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Headers:{headers}")
169	1
170	1	methods = self.tabpy_state.get_access_control_allow_methods()
171		if len(methods) > 0:
172		self.set_header("Access-Control-Allow-Methods", methods)
173		self.logger.log(logging.DEBUG, f"Access-Control-Allow-Methods:{methods}")
174	1
175		def _get_auth_method(self, api_version) -> (bool, str):
176		"""
177		Finds authentication method if provided.
178
179		Parameters
180		----------
181		api_version : str
182		API version for authentication.
183
184		Returns
185		-------
186		bool
187		True if known authentication method is found.
188		False otherwise.
189
190		str
191		Name of authentication method used by client.
192		If empty no authentication required.
193
194		(True, '') as result of this function means authentication
195		is not needed.
196	1	"""
197		if api_version not in self.settings[SettingsParameters.ApiVersions]:
198		self.logger.log(logging.CRITICAL, f'Unknown API version "{api_version}"')
199		return False, ""
200	1
201	1	version_settings = self.settings[SettingsParameters.ApiVersions][api_version]
202		if "features" not in version_settings:
203		self.logger.log(
204		logging.INFO, f'No features configured for API "{api_version}"'
205		)
206		return True, ""
207	1
208	1	features = version_settings["features"]
209		if (
210		"authentication" not in features
211		or not features["authentication"]["required"]
212	1	):
213		self.logger.log(
214		logging.INFO,
215		"Authentication is not a required feature for API " f'"{api_version}"',
216	1	)
217		return True, ""
218	1
219	1	auth_feature = features["authentication"]
220		if "methods" not in auth_feature:
221		self.logger.log(
222		logging.INFO,
223		"Authentication method is not configured for API " f'"{api_version}"',
224		)
225	1
226	1	methods = auth_feature["methods"]
227	1	if "basic-auth" in auth_feature["methods"]:
228		return True, "basic-auth"
229		# Add new methods here...
230
231		# No known methods were found
232		self.logger.log(
233		logging.CRITICAL,
234		f'Unknown authentication method(s) "{methods}" are configured '
235		f'for API "{api_version}"',
236		)
237		return False, ""
238	1
239		def _get_basic_auth_credentials(self) -> bool:
240		"""
241		Find credentials for basic access authentication method. Credentials if
242		found stored in Credentials.username and Credentials.password.
243
244		Returns
245		-------
246		bool
247		True if valid credentials were found.
248		False otherwise.
249	1	"""
250		self.logger.log(
251		logging.DEBUG, "Checking request headers for authentication data"
252	1	)
253	1	if "Authorization" not in self.request.headers:
254	1	self.logger.log(logging.INFO, "Authorization header not found")
255		return False
256	1
257	1	auth_header = self.request.headers["Authorization"]
258	1	auth_header_list = auth_header.split(" ")
259		if len(auth_header_list) != 2 or auth_header_list[0] != "Basic":
260		self.logger.log(
261		logging.ERROR, f'Unknown authentication method "{auth_header}"'
262		)
263		return False
264	1
265	1	try:
266		cred = base64.b64decode(auth_header_list[1]).decode("utf-8")
267		except (binascii.Error, UnicodeDecodeError) as ex:
268		self.logger.log(logging.CRITICAL, f"Cannot decode credentials: {str(ex)}")
269		return False
270	1
271	1	login_pwd = cred.split(":")
272		if len(login_pwd) != 2:
273		self.logger.log(logging.ERROR, "Invalid string in encoded credentials")
274		return False
275	1
276	1	self.username = login_pwd[0]
277	1	self.logger.set_tabpy_username(self.username)
278	1	self.password = login_pwd[1]
279		return True
280	1
281		def _get_credentials(self, method) -> bool:
282		"""
283		Find credentials for specified authentication method. Credentials if
284		found stored in self.username and self.password.
285
286		Parameters
287		----------
288		method: str
289		Authentication method name.
290
291		Returns
292		-------
293		bool
294		True if valid credentials were found.
295		False otherwise.
296	1	"""
297	1	if method == "basic-auth":
298		return self._get_basic_auth_credentials()
299		# Add new methods here...
300
301		# No known methods were found
302		self.logger.log(
303		logging.CRITICAL,
304		f'Unknown authentication method(s) "{method}" are configured ',
305		)
306		return False
307	1
308		def _validate_basic_auth_credentials(self) -> bool:
309		"""
310		Validates username:pwd if they are the same as
311		stored credentials.
312
313		Returns
314		-------
315		bool
316		True if credentials has key login and
317		credentials[login] equal SHA3(pwd), False
318		otherwise.
319	1	"""
320	1	login = self.username.lower()
321		self.logger.log(
322		logging.DEBUG, f'Validating credentials for user name "{login}"'
323	1	)
324	1	if login not in self.credentials:
325	1	self.logger.log(logging.ERROR, f'User name "{self.username}" not found')
326		return False
327	1
328	1	hashed_pwd = hash_password(login, self.password)
329		if self.credentials[login].lower() != hashed_pwd.lower():
330		self.logger.log(
331		logging.ERROR, f'Wrong password for user name "{self.username}"'
332		)
333		return False
334	1
335		return True
336	1
337		def _validate_credentials(self, method) -> bool:
338		"""
339		Validates credentials according to specified methods if they
340		are what expected.
341
342		Parameters
343		----------
344		method: str
345		Authentication method name.
346
347		Returns
348		-------
349		bool
350		True if credentials are valid.
351		False otherwise.
352	1	"""
353	1	if method == "basic-auth":
354		return self._validate_basic_auth_credentials()
355		# Add new methods here...
356
357		# No known methods were found
358		self.logger.log(
359		logging.CRITICAL,
360		f'Unknown authentication method(s) "{method}" are configured ',
361		)
362		return False
363	1
364		def handle_authentication(self, api_version):
365		"""
366		If authentication feature is configured checks provided
367		credentials.
368
369		Parameters
370		----------
371		api_version : str
372		API version for authentication.
373
374		Returns
375		-------
376		String
377		None if authentication is not required and username and password are None.
378		None if authentication is required and valid credentials provided.
379		NotAuthorized if authenication is required and credentials are incorrect.
380		NotRequired if authentication is not required but credentials are provided.
381	1	"""
382	1	self.logger.log(logging.DEBUG, "Handling authentication")
383	1	found, method = self._get_auth_method(api_version)
384		if not found:
385		return AuthErrorStates.NotAuthorized
386	1
387	1	if method == "":
388	1	if not self._get_basic_auth_credentials():
389		self.logger.log(logging.DEBUG,
390	1	"authentication not required, username and password are none")
391		return AuthErrorStates.NONE
392	1	else:
393		self.logger.log(logging.DEBUG,
394	1	"authentication not required, username and password are not none")
395		return AuthErrorStates.NotRequired
396	1
397	1	if not self._get_credentials(method):
398		return AuthErrorStates.NotAuthorized
399	1
400	1	if not self._validate_credentials(method):
401		return AuthErrorStates.NotAuthorized
402	1
403		return AuthErrorStates.NONE
404	1
405		def should_fail_with_auth_error(self):
406		"""
407		Checks if authentication is required:
408		- if it is not returns false, None
409		- if it is required validates provided credentials
410
411		Returns
412		-------
413		bool
414		False if authentication is not required and username
415		and password is None or isrequired and validation
416		for credentials passes.
417		True if validation for credentials failed or
418		if authentication is not required and username and password
419		fields are not empty.
420	1	"""
421		return self.auth_error
422	1
423		def fail_with_auth_error(self):
424		"""
425		Prepares server 401 response and server 406 response depending
426		on the value of the self.auth_error flag
427	1	"""
428	1	if self.auth_error == AuthErrorStates.NotAuthorized:
429	1	self.logger.log(logging.ERROR, "Failing with 401 for unauthorized request")
430	1	self.set_status(401)
431	1	self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
432		self.error_out(
433		401,
434		info="Unauthorized request.",
435		log_message="Invalid credentials provided.",
436		)
437	1	else:
438	1	self.logger.log(logging.ERROR, "Failing with 406 for Not Acceptable")
439	1	self.set_status(406)
440	1	self.set_header("WWW-Authenticate", f'Basic realm="{self.tabpy_state.name}"')
441		self.error_out(
442		406,
443		info="Not Acceptable",
444		log_message="Username or password provided when authentication not available.",
445		)
446

tableau / TabPy

Pull Request — master (#566)

tabpy.tabpy_server.handlers.base_handler.BaseHandler._get_auth_method() B

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like