Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/oauthlib/oauth2/rfc6749/endpoints/base.py: 38%

1""" 

2oauthlib.oauth2.rfc6749 

3~~~~~~~~~~~~~~~~~~~~~~~ 

4 

5This module is an implementation of various logic needed 

6for consuming and providing OAuth 2.0 RFC6749. 

7""" 

8import functools 

9import logging 

10 

11from ..errors import ( 

12 FatalClientError, InvalidClientError, InvalidRequestError, OAuth2Error, 

13 ServerError, TemporarilyUnavailableError, UnsupportedTokenTypeError, 

14) 

15 

16log = logging.getLogger(__name__) 

17 

18 

19class BaseEndpoint: 

20 

21 def __init__(self): 

22 self._available = True 

23 self._catch_errors = False 

24 self._valid_request_methods = None 

25 

26 @property 

27 def valid_request_methods(self): 

28 return self._valid_request_methods 

29 

30 @valid_request_methods.setter 

31 def valid_request_methods(self, valid_request_methods): 

32 if valid_request_methods is not None: 

33 valid_request_methods = [x.upper() for x in valid_request_methods] 

34 self._valid_request_methods = valid_request_methods 

35 

36 

37 @property 

38 def available(self): 

39 return self._available 

40 

41 @available.setter 

42 def available(self, available): 

43 self._available = available 

44 

45 @property 

46 def catch_errors(self): 

47 return self._catch_errors 

48 

49 @catch_errors.setter 

50 def catch_errors(self, catch_errors): 

51 self._catch_errors = catch_errors 

52 

53 def _raise_on_missing_token(self, request): 

54 """Raise error on missing token.""" 

55 if not request.token: 

56 raise InvalidRequestError(request=request, 

57 description='Missing token parameter.') 

58 def _raise_on_invalid_client(self, request): 

59 """Raise on failed client authentication.""" 

60 if self.request_validator.client_authentication_required(request): 

61 if not self.request_validator.authenticate_client(request): 

62 log.debug('Client authentication failed, %r.', request) 

63 raise InvalidClientError(request=request) 

64 elif not self.request_validator.authenticate_client_id(request.client_id, request): 

65 log.debug('Client authentication failed, %r.', request) 

66 raise InvalidClientError(request=request) 

67 

68 def _raise_on_unsupported_token(self, request): 

69 """Raise on unsupported tokens.""" 

70 if (request.token_type_hint and 

71 request.token_type_hint in self.valid_token_types and 

72 request.token_type_hint not in self.supported_token_types): 

73 raise UnsupportedTokenTypeError(request=request) 

74 

75 def _raise_on_bad_method(self, request): 

76 if self.valid_request_methods is None: 

77 raise ValueError('Configure "valid_request_methods" property first') 

78 if request.http_method.upper() not in self.valid_request_methods: 

79 raise InvalidRequestError(request=request, 

80 description=('Unsupported request method %s' % request.http_method.upper())) 

81 

82 def _raise_on_bad_post_request(self, request): 

83 """Raise if invalid POST request received 

84 """ 

85 if request.http_method.upper() == 'POST': 

86 query_params = request.uri_query or "" 

87 if query_params: 

88 raise InvalidRequestError(request=request, 

89 description=('URL query parameters are not allowed')) 

90 

91def catch_errors_and_unavailability(f): 

92 @functools.wraps(f) 

93 def wrapper(endpoint, uri, *args, **kwargs): 

94 if not endpoint.available: 

95 e = TemporarilyUnavailableError() 

96 log.info('Endpoint unavailable, ignoring request %s.' % uri) 

97 return {}, e.json, 503 

98 

99 if endpoint.catch_errors: 

100 try: 

101 return f(endpoint, uri, *args, **kwargs) 

102 except OAuth2Error: 

103 raise 

104 except FatalClientError: 

105 raise 

106 except Exception as e: 

107 error = ServerError() 

108 log.warning( 

109 'Exception caught while processing request, %s.' % e) 

110 return {}, error.json, 500 

111 else: 

112 return f(endpoint, uri, *args, **kwargs) 

113 return wrapper