tabpy.tabpy_server.handlers.base_handler.BaseHandler.options() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

BaseHandler.options() A
last analyzed 2024-11-25 22:00 UTC

↳ 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.max_request_size = app.max_request_size

        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.",
            )

    def request_body_size_within_limit(self):
        """
        Determines if the request body size is within the specified limit.
        
        Returns
        -------
        bool
            True if the request body size is within the limit, False otherwise.
        """
        if self.max_request_size is not None:
            if "Content-Length" in self.request.headers:
                content_length = int(self.request.headers["Content-Length"])
                if content_length > self.max_request_size:
                    self.error_out(
                        413,
                        info="Request Entity Too Large",
                        log_message=f"Request with size {content_length} exceeded limit of {self.max_request_size} (bytes).",
                    )
                    return False

        return True


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	1	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	1	self.max_request_size = app.max_request_size
131
132	1	self.logger = ContextLoggerWrapper(self.request)
133	1	self.logger.enable_context_logging(
134		app.settings[SettingsParameters.LogRequestContext]
135		)
136	1	self.logger.log(logging.DEBUG, "Checking if need to handle authentication")
137	1	self.auth_error = self.handle_authentication("v1")
138
139	1	def error_out(self, code, log_message, info=None):
140	1	self.set_status(code)
141	1	self.write(json.dumps({"message": log_message, "info": info or {}}))
142
143	1	self.logger.log(
144		logging.ERROR,
145		'Responding with status={}, message="{}", info="{}"'.format(
146		code, log_message, info
147		),
148		)
149
150	1	def options(self):
151		# add CORS headers if TabPy has a cors_origin specified
152		self._add_CORS_header()
153		self.write({})
154
155	1	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		"""
160	1	origin = self.tabpy_state.get_access_control_allow_origin()
161	1	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
165	1	headers = self.tabpy_state.get_access_control_allow_headers()
166	1	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
170	1	methods = self.tabpy_state.get_access_control_allow_methods()
171	1	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
175	1	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		"""
197	1	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
201	1	version_settings = self.settings[SettingsParameters.ApiVersions][api_version]
202	1	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
208	1	features = version_settings["features"]
209	1	if (
210		"authentication" not in features
211		or not features["authentication"]["required"]
212		):
213	1	self.logger.log(
214		logging.INFO,
215		"Authentication is not a required feature for API " f'"{api_version}"',
216		)
217	1	return True, ""
218
219	1	auth_feature = features["authentication"]
220	1	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
226	1	methods = auth_feature["methods"]
227	1	if "basic-auth" in auth_feature["methods"]:
228	1	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
239	1	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		"""
250	1	self.logger.log(
251		logging.DEBUG, "Checking request headers for authentication data"
252		)
253	1	if "Authorization" not in self.request.headers:
254	1	self.logger.log(logging.INFO, "Authorization header not found")
255	1	return False
256
257	1	auth_header = self.request.headers["Authorization"]
258	1	auth_header_list = auth_header.split(" ")
259	1	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
265	1	try:
266	1	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
271	1	login_pwd = cred.split(":")
272	1	if len(login_pwd) != 2:
273		self.logger.log(logging.ERROR, "Invalid string in encoded credentials")
274		return False
275
276	1	self.username = login_pwd[0]
277	1	self.logger.set_tabpy_username(self.username)
278	1	self.password = login_pwd[1]
279	1	return True
280
281	1	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		"""
297	1	if method == "basic-auth":
298	1	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
308	1	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		"""
320	1	login = self.username.lower()
321	1	self.logger.log(
322		logging.DEBUG, f'Validating credentials for user name "{login}"'
323		)
324	1	if login not in self.credentials:
325	1	self.logger.log(logging.ERROR, f'User name "{self.username}" not found')
326	1	return False
327
328	1	hashed_pwd = hash_password(login, self.password)
329	1	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
335	1	return True
336
337	1	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		"""
353	1	if method == "basic-auth":
354	1	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
364	1	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		"""
382	1	self.logger.log(logging.DEBUG, "Handling authentication")
383	1	found, method = self._get_auth_method(api_version)
384	1	if not found:
385		return AuthErrorStates.NotAuthorized
386
387	1	if method == "":
388	1	if not self._get_basic_auth_credentials():
389	1	self.logger.log(logging.DEBUG,
390		"authentication not required, username and password are none")
391	1	return AuthErrorStates.NONE
392		else:
393	1	self.logger.log(logging.DEBUG,
394		"authentication not required, username and password are not none")
395	1	return AuthErrorStates.NotRequired
396
397	1	if not self._get_credentials(method):
398	1	return AuthErrorStates.NotAuthorized
399
400	1	if not self._validate_credentials(method):
401	1	return AuthErrorStates.NotAuthorized
402
403	1	return AuthErrorStates.NONE
404
405	1	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		"""
421	1	return self.auth_error
422
423	1	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		"""
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	1	self.error_out(
433		401,
434		info="Unauthorized request.",
435		log_message="Invalid credentials provided.",
436		)
437		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	1	self.error_out(
442		406,
443		info="Not Acceptable",
444		log_message="Username or password provided when authentication not available.",
445		)
446
447	1	def request_body_size_within_limit(self):
448		"""
449		Determines if the request body size is within the specified limit.
450
451		Returns
452		-------
453		bool
454		True if the request body size is within the limit, False otherwise.
455		"""
456	1	if self.max_request_size is not None:
457	1	if "Content-Length" in self.request.headers:
458	1	content_length = int(self.request.headers["Content-Length"])
459	1	if content_length > self.max_request_size:
460	1	self.error_out(
461		413,
462		info="Request Entity Too Large",
463		log_message=f"Request with size {content_length} exceeded limit of {self.max_request_size} (bytes).",
464		)
465	1	return False
466
467		return True
468

tableau / TabPy

BaseHandler.options() A last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

BaseHandler.options() A
last analyzed 2024-11-25 22:00 UTC