Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sigstore/_internal/rekor/checkpoint.py: 40%

1# Copyright 2023 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""" 

16Rekor Checkpoint machinery. 

17""" 

18 

19from __future__ import annotations 

20 

21import base64 

22import binascii 

23import re 

24import struct 

25import typing 

26from dataclasses import dataclass 

27 

28from pydantic import BaseModel, Field, StrictStr 

29 

30from sigstore._utils import KeyID 

31from sigstore.errors import VerificationError 

32 

33if typing.TYPE_CHECKING: 

34 from sigstore._internal.trust import RekorKeyring 

35 from sigstore.models import TransparencyLogEntry 

36 

37 

38@dataclass(frozen=True) 

39class RekorSignature: 

40 """ 

41 Represents a `RekorSignature` containing: 

42 

43 - the name of the signature, e.g. "rekor.sigstage.dev" 

44 - the signature hash 

45 - the base64 signature 

46 """ 

47 

48 name: str 

49 sig_hash: bytes 

50 signature: bytes 

51 

52 

53class LogCheckpoint(BaseModel): 

54 """ 

55 Represents a Rekor `LogCheckpoint` containing: 

56 

57 - an origin, e.g. "rekor.sigstage.dev - 8050909264565447525" 

58 - the size of the log, 

59 - the hash of the log, 

60 - and any optional ancillary constants, e.g. "Timestamp: 1679349379012118479" 

61 

62 See: <https://github.com/transparency-dev/formats/blob/main/log/README.md> 

63 """ 

64 

65 origin: StrictStr 

66 log_size: int 

67 log_hash: StrictStr 

68 other_content: list[str] 

69 

70 @classmethod 

71 def from_text(cls, text: str) -> LogCheckpoint: 

72 """ 

73 Serialize from the text header ("note") of a SignedNote. 

74 """ 

75 

76 lines = text.strip().split("\n") 

77 if len(lines) < 3: 

78 raise VerificationError("malformed LogCheckpoint: too few items in header") 

79 

80 origin = lines[0] 

81 if len(origin) == 0: 

82 raise VerificationError("malformed LogCheckpoint: empty origin") 

83 

84 try: 

85 log_size = int(lines[1]) 

86 except ValueError: 

87 raise VerificationError("malformed LogCheckpoint: invalid log size") 

88 

89 try: 

90 root_hash = base64.b64decode(lines[2]).hex() 

91 except binascii.Error: 

92 raise VerificationError("malformed LogCheckpoint: invalid root hash") 

93 

94 return LogCheckpoint( 

95 origin=origin, 

96 log_size=log_size, 

97 log_hash=root_hash, 

98 other_content=lines[3:], 

99 ) 

100 

101 @classmethod 

102 def to_text(self) -> str: 

103 """ 

104 Serialize a `LogCheckpoint` into text format. 

105 See class definition for a prose description of the format. 

106 """ 

107 return "\n".join( 

108 [self.origin, str(self.log_size), self.log_hash, *self.other_content] 

109 ) 

110 

111 

112@dataclass(frozen=True) 

113class SignedNote: 

114 """ 

115 Represents a "signed note" containing a note and its corresponding list of signatures. 

116 """ 

117 

118 note: StrictStr = Field(..., alias="note") 

119 signatures: list[RekorSignature] = Field(..., alias="signatures") 

120 

121 @classmethod 

122 def from_text(cls, text: str) -> SignedNote: 

123 """ 

124 Deserialize from a bundled text 'note'. 

125 

126 A note contains: 

127 - a name, a string associated with the signer, 

128 - a separator blank line, 

129 - and signature(s), each signature takes the form 

130 `\u2014 NAME SIGNATURE\n` 

131 (where \u2014 == em dash). 

132 

133 This is derived from Rekor's `UnmarshalText`: 

134 <https://github.com/sigstore/rekor/blob/4b1fa6661cc6dfbc844b4c6ed9b1f44e7c5ae1c0/pkg/util/signed_note.go#L141> 

135 """ 

136 

137 separator: str = "\n\n" 

138 if text.count(separator) != 1: 

139 raise VerificationError( 

140 "note must contain one blank line, delineating the text from the signature block" 

141 ) 

142 split = text.index(separator) 

143 

144 header: str = text[: split + 1] 

145 data: str = text[split + len(separator) :] 

146 

147 if len(data) == 0: 

148 raise VerificationError( 

149 "malformed Note: must contain at least one signature" 

150 ) 

151 if data[-1] != "\n": 

152 raise VerificationError( 

153 "malformed Note: data section must end with newline" 

154 ) 

155 

156 sig_parser = re.compile(r"\u2014 (\S+) (\S+)\n") 

157 signatures: list[RekorSignature] = [] 

158 for name, signature in re.findall(sig_parser, data): 

159 signature_bytes: bytes = base64.b64decode(signature) 

160 if len(signature_bytes) < 5: 

161 raise VerificationError( 

162 "malformed Note: signature contains too few bytes" 

163 ) 

164 

165 signature = RekorSignature( 

166 name=name, 

167 sig_hash=struct.unpack(">4s", signature_bytes[0:4])[0], 

168 signature=base64.b64encode(signature_bytes[4:]), 

169 ) 

170 signatures.append(signature) 

171 

172 return cls(note=header, signatures=signatures) 

173 

174 def verify(self, rekor_keyring: RekorKeyring, key_id: KeyID) -> None: 

175 """ 

176 Verify the `SignedNote` using the given RekorKeyring and KeyID. 

177 """ 

178 

179 note = str.encode(self.note) 

180 

181 for sig in self.signatures: 

182 if sig.sig_hash == key_id[:4]: 

183 try: 

184 rekor_keyring.verify( 

185 key_id=key_id, 

186 signature=base64.b64decode(sig.signature), 

187 data=note, 

188 ) 

189 return 

190 except VerificationError as sig_err: 

191 raise VerificationError(f"checkpoint: invalid signature: {sig_err}") 

192 

193 raise VerificationError( 

194 f"checkpoint: Signature not found for log ID {key_id.hex()}" 

195 ) 

196 

197 

198@dataclass(frozen=True) 

199class SignedCheckpoint: 

200 """ 

201 Represents a *signed* `Checkpoint`: a `LogCheckpoint` and its corresponding `SignedNote`. 

202 """ 

203 

204 signed_note: SignedNote 

205 checkpoint: LogCheckpoint 

206 

207 @classmethod 

208 def from_text(cls, text: str) -> SignedCheckpoint: 

209 """ 

210 Create a new `SignedCheckpoint` from the text representation. 

211 """ 

212 

213 signed_note = SignedNote.from_text(text) 

214 checkpoint = LogCheckpoint.from_text(signed_note.note) 

215 return cls(signed_note=signed_note, checkpoint=checkpoint) 

216 

217 

218def verify_checkpoint(rekor_keyring: RekorKeyring, entry: TransparencyLogEntry) -> None: 

219 """ 

220 Verify the inclusion proof's checkpoint. 

221 """ 

222 

223 inclusion_proof = entry._inner.inclusion_proof 

224 if inclusion_proof.checkpoint is None: 

225 raise VerificationError("Inclusion proof does not contain a checkpoint") 

226 

227 # verification occurs in two stages: 

228 # 1) verify the signature on the checkpoint 

229 # 2) verify the root hash in the checkpoint matches the root hash from the inclusion proof. 

230 signed_checkpoint = SignedCheckpoint.from_text(inclusion_proof.checkpoint.envelope) 

231 signed_checkpoint.signed_note.verify( 

232 rekor_keyring, 

233 KeyID(entry._inner.log_id.key_id), 

234 ) 

235 

236 checkpoint_hash = signed_checkpoint.checkpoint.log_hash 

237 root_hash = inclusion_proof.root_hash.hex() 

238 

239 if checkpoint_hash != root_hash: 

240 raise VerificationError( 

241 "Inclusion proof contains invalid root hash signature: ", 

242 f"expected {checkpoint_hash} got {root_hash}", 

243 )