Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/stripe/webhook.py: 38%

1from __future__ import absolute_import, division, print_function 

2 

3import hmac 

4import json 

5import time 

6from collections import OrderedDict 

7from hashlib import sha256 

8 

9import stripe 

10from stripe import error, util 

11 

12 

13class Webhook(object): 

14 DEFAULT_TOLERANCE = 300 

15 

16 @staticmethod 

17 def construct_event( 

18 payload, sig_header, secret, tolerance=DEFAULT_TOLERANCE, api_key=None 

19 ): 

20 if hasattr(payload, "decode"): 

21 payload = payload.decode("utf-8") 

22 

23 WebhookSignature.verify_header(payload, sig_header, secret, tolerance) 

24 

25 data = json.loads(payload, object_pairs_hook=OrderedDict) 

26 event = stripe.Event.construct_from(data, api_key or stripe.api_key) 

27 

28 return event 

29 

30 

31class WebhookSignature(object): 

32 EXPECTED_SCHEME = "v1" 

33 

34 @staticmethod 

35 def _compute_signature(payload, secret): 

36 mac = hmac.new( 

37 secret.encode("utf-8"), 

38 msg=payload.encode("utf-8"), 

39 digestmod=sha256, 

40 ) 

41 return mac.hexdigest() 

42 

43 @staticmethod 

44 def _get_timestamp_and_signatures(header, scheme): 

45 list_items = [i.split("=", 2) for i in header.split(",")] 

46 timestamp = int([i[1] for i in list_items if i[0] == "t"][0]) 

47 signatures = [i[1] for i in list_items if i[0] == scheme] 

48 return timestamp, signatures 

49 

50 @classmethod 

51 def verify_header(cls, payload, header, secret, tolerance=None): 

52 try: 

53 timestamp, signatures = cls._get_timestamp_and_signatures( 

54 header, cls.EXPECTED_SCHEME 

55 ) 

56 except Exception: 

57 raise error.SignatureVerificationError( 

58 "Unable to extract timestamp and signatures from header", 

59 header, 

60 payload, 

61 ) 

62 

63 if not signatures: 

64 raise error.SignatureVerificationError( 

65 "No signatures found with expected scheme " 

66 "%s" % cls.EXPECTED_SCHEME, 

67 header, 

68 payload, 

69 ) 

70 

71 signed_payload = "%d.%s" % (timestamp, payload) 

72 expected_sig = cls._compute_signature(signed_payload, secret) 

73 if not any(util.secure_compare(expected_sig, s) for s in signatures): 

74 raise error.SignatureVerificationError( 

75 "No signatures found matching the expected signature for " 

76 "payload", 

77 header, 

78 payload, 

79 ) 

80 

81 if tolerance and timestamp < time.time() - tolerance: 

82 raise error.SignatureVerificationError( 

83 "Timestamp outside the tolerance zone (%d)" % timestamp, 

84 header, 

85 payload, 

86 ) 

87 

88 return True