Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/requests_oauthlib/oauth1_auth.py: 32%

1# -*- coding: utf-8 -*- 

2import logging 

3 

4from oauthlib.common import extract_params 

5from oauthlib.oauth1 import Client, SIGNATURE_HMAC, SIGNATURE_TYPE_AUTH_HEADER 

6from oauthlib.oauth1 import SIGNATURE_TYPE_BODY 

7from requests.utils import to_native_string 

8from requests.auth import AuthBase 

9 

10CONTENT_TYPE_FORM_URLENCODED = "application/x-www-form-urlencoded" 

11CONTENT_TYPE_MULTI_PART = "multipart/form-data" 

12 

13 

14log = logging.getLogger(__name__) 

15 

16# OBS!: Correct signing of requests are conditional on invoking OAuth1 

17# as the last step of preparing a request, or at least having the 

18# content-type set properly. 

19class OAuth1(AuthBase): 

20 """Signs the request using OAuth 1 (RFC5849)""" 

21 

22 client_class = Client 

23 

24 def __init__( 

25 self, 

26 client_key, 

27 client_secret=None, 

28 resource_owner_key=None, 

29 resource_owner_secret=None, 

30 callback_uri=None, 

31 signature_method=SIGNATURE_HMAC, 

32 signature_type=SIGNATURE_TYPE_AUTH_HEADER, 

33 rsa_key=None, 

34 verifier=None, 

35 decoding="utf-8", 

36 client_class=None, 

37 force_include_body=False, 

38 **kwargs 

39 ): 

40 

41 try: 

42 signature_type = signature_type.upper() 

43 except AttributeError: 

44 pass 

45 

46 client_class = client_class or self.client_class 

47 

48 self.force_include_body = force_include_body 

49 

50 self.client = client_class( 

51 client_key, 

52 client_secret, 

53 resource_owner_key, 

54 resource_owner_secret, 

55 callback_uri, 

56 signature_method, 

57 signature_type, 

58 rsa_key, 

59 verifier, 

60 decoding=decoding, 

61 **kwargs 

62 ) 

63 

64 def __call__(self, r): 

65 """Add OAuth parameters to the request. 

66 

67 Parameters may be included from the body if the content-type is 

68 urlencoded, if no content type is set a guess is made. 

69 """ 

70 # Overwriting url is safe here as request will not modify it past 

71 # this point. 

72 log.debug("Signing request %s using client %s", r, self.client) 

73 

74 content_type = r.headers.get("Content-Type", "") 

75 if ( 

76 not content_type 

77 and extract_params(r.body) 

78 or self.client.signature_type == SIGNATURE_TYPE_BODY 

79 ): 

80 content_type = CONTENT_TYPE_FORM_URLENCODED 

81 if not isinstance(content_type, str): 

82 content_type = content_type.decode("utf-8") 

83 

84 is_form_encoded = CONTENT_TYPE_FORM_URLENCODED in content_type 

85 

86 log.debug( 

87 "Including body in call to sign: %s", 

88 is_form_encoded or self.force_include_body, 

89 ) 

90 

91 if is_form_encoded: 

92 r.headers["Content-Type"] = CONTENT_TYPE_FORM_URLENCODED 

93 r.url, headers, r.body = self.client.sign( 

94 str(r.url), str(r.method), r.body or "", r.headers 

95 ) 

96 elif self.force_include_body: 

97 # To allow custom clients to work on non form encoded bodies. 

98 r.url, headers, r.body = self.client.sign( 

99 str(r.url), str(r.method), r.body or "", r.headers 

100 ) 

101 else: 

102 # Omit body data in the signing of non form-encoded requests 

103 r.url, headers, _ = self.client.sign( 

104 str(r.url), str(r.method), None, r.headers 

105 ) 

106 

107 r.prepare_headers(headers) 

108 r.url = to_native_string(r.url) 

109 log.debug("Updated url: %s", r.url) 

110 log.debug("Updated headers: %s", headers) 

111 log.debug("Updated body: %r", r.body) 

112 return r