Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sigstore/_internal/timestamp.py: 49%

1# Copyright 2022 The Sigstore Authors 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15""" 

16Utilities to deal with sources of signed time. 

17""" 

18 

19import enum 

20import threading 

21from dataclasses import dataclass 

22from datetime import datetime 

23 

24import requests 

25from rfc3161_client import ( 

26 TimestampRequestBuilder, 

27 TimeStampResponse, 

28 decode_timestamp_response, 

29) 

30from rfc3161_client.base import HashAlgorithm 

31 

32from sigstore._internal import USER_AGENT 

33 

34CLIENT_TIMEOUT: int = 5 

35 

36 

37class TimestampSource(enum.Enum): 

38 """Represents the source of a timestamp.""" 

39 

40 TIMESTAMP_AUTHORITY = enum.auto() 

41 TRANSPARENCY_SERVICE = enum.auto() 

42 

43 

44@dataclass 

45class TimestampVerificationResult: 

46 """Represents a timestamp used by the Verifier. 

47 

48 A Timestamp either comes from a Timestamping Service (RFC3161) or the Transparency 

49 Service. 

50 """ 

51 

52 source: TimestampSource 

53 time: datetime 

54 

55 

56class TimestampError(Exception): 

57 """ 

58 A generic error in the TimestampAuthority client. 

59 """ 

60 

61 pass 

62 

63 

64class TimestampAuthorityClient: 

65 """Internal client to deal with a Timestamp Authority""" 

66 

67 def __init__(self, url: str) -> None: 

68 """ 

69 Create a new `TimestampAuthorityClient` from the given URL. 

70 """ 

71 self.url = url 

72 self._thread_local = threading.local() 

73 

74 @property 

75 def _session(self) -> requests.Session: 

76 """ 

77 Lazy-initialized thread-local session object 

78 """ 

79 if not hasattr(self._thread_local, "session"): 

80 session = requests.Session() 

81 session.headers.update( 

82 { 

83 "Content-Type": "application/timestamp-query", 

84 "User-Agent": USER_AGENT, 

85 } 

86 ) 

87 self._thread_local.session = session 

88 return self._thread_local.session # type: ignore[no-any-return] 

89 

90 def request_timestamp(self, signature: bytes) -> TimeStampResponse: 

91 """ 

92 Timestamp the signature using the configured Timestamp Authority. 

93 

94 This method generates a RFC3161 Timestamp Request and sends it to a TSA. 

95 The received response is parsed but *not* cryptographically verified. 

96 

97 request_timestamp() can be called from multiple threads. 

98 

99 Raises a TimestampError on failure. 

100 """ 

101 # Build the timestamp request 

102 try: 

103 timestamp_request = ( 

104 TimestampRequestBuilder() 

105 .hash_algorithm(HashAlgorithm.SHA256) 

106 .data(signature) 

107 .nonce(nonce=True) 

108 .build() 

109 ) 

110 except ValueError as error: 

111 msg = f"invalid request: {error}" 

112 raise TimestampError(msg) 

113 

114 # Send it to the TSA for signing 

115 try: 

116 response = self._session.post( 

117 self.url, 

118 data=timestamp_request.as_bytes(), 

119 timeout=CLIENT_TIMEOUT, 

120 ) 

121 response.raise_for_status() 

122 except requests.RequestException as error: 

123 msg = f"error while sending the request to the TSA: {error}" 

124 raise TimestampError(msg) 

125 

126 # Check that we can parse the response but do not *verify* it 

127 try: 

128 timestamp_response = decode_timestamp_response(response.content) 

129 except ValueError as e: 

130 msg = f"invalid response: {e}" 

131 raise TimestampError(msg) 

132 

133 return timestamp_response