11from __future__ import annotations
22
33import argparse
4+ import hashlib
5+ import json
6+ import re
47import subprocess
8+ from dataclasses import dataclass
59from pathlib import Path
10+ from typing import Any
611
712
13+ REVIEW_PATH = Path ("docs/security-review/rc2-external-review.json" )
14+ WINDOW_PATH = Path ("docs/rc-window/1.0.0rc2.json" )
815EXPECTED_RECORD_CHANGES = (
9- ("A" , "docs/rc-window/1.0.0rc2.json" ),
10- ("M" , "docs/security-review/rc2-external-review.json" ),
16+ ("A" , str ( WINDOW_PATH ) ),
17+ ("M" , str ( REVIEW_PATH ) ),
1118)
19+ FULL_SHA = re .compile (r"^[0-9a-f]{40}$" )
20+ PENDING_REVIEW : dict [str , Any ] = {
21+ "schema_version" : "govengine.rc2_external_security_review.v1" ,
22+ "source_commit" : "" ,
23+ "artifacts" : {
24+ "runner" : "github-hosted-runner" ,
25+ "wheel_sha256" : "" ,
26+ "normalized_sdist_sha256" : "" ,
27+ },
28+ "confidential_report_sha256" : "" ,
29+ "reviewer" : "" ,
30+ "reviewed_at" : None ,
31+ "verdict" : "pending_external_reviewer" ,
32+ "open_p0" : None ,
33+ "open_p1" : None ,
34+ }
35+
36+
37+ @dataclass (frozen = True )
38+ class ReleaseABState :
39+ mode : str
40+ source_commit : str
41+ record_commit : str | None
1242
1343
1444def _git (repo : Path , * args : str ) -> str :
1545 return subprocess .check_output (["git" , * args ], cwd = repo , text = True ).strip ()
1646
1747
48+ def _git_bytes (repo : Path , * args : str ) -> bytes :
49+ return subprocess .check_output (["git" , * args ], cwd = repo )
50+
51+
52+ def _load_json (text : str ) -> Any :
53+ def reject_duplicate_keys (pairs : list [tuple [str , Any ]]) -> dict [str , Any ]:
54+ value : dict [str , Any ] = {}
55+ for key , item in pairs :
56+ if key in value :
57+ raise ValueError (f"duplicate JSON key:{ key } )
58+ value [key ] = item
59+ return value
60+
61+ return json .loads (text , object_pairs_hook = reject_duplicate_keys )
62+
63+
1864def validate_record_commit (repo : Path , review_commit : str ) -> str :
1965 parents = _git (repo , "rev-list" , "--parents" , "-n" , "1" , review_commit ).split ()
2066 if len (parents ) != 2 :
@@ -29,14 +75,157 @@ def validate_record_commit(repo: Path, review_commit: str) -> str:
2975 return source_commit
3076
3177
78+ def _matches_authentic_record (
79+ repo : Path ,
80+ commit : str ,
81+ source_commit : str ,
82+ current_review : bytes ,
83+ ) -> bool :
84+ try :
85+ if validate_record_commit (repo , commit ) != source_commit :
86+ return False
87+ record_review = _git_bytes (repo , "show" , f"{ commit } { REVIEW_PATH } )
88+ record_window = _load_json (
89+ _git_bytes (repo , "show" , f"{ commit } { WINDOW_PATH } ).decode ("utf-8" )
90+ )
91+ except (ValueError , subprocess .CalledProcessError , json .JSONDecodeError ):
92+ return False
93+ record_reference = (
94+ record_window .get ("security_review" )
95+ if isinstance (record_window , dict )
96+ else None
97+ )
98+ return (
99+ record_review == current_review
100+ and isinstance (record_reference , dict )
101+ and record_window .get ("source_commit" ) == source_commit
102+ and record_reference .get ("path" ) == str (REVIEW_PATH )
103+ and record_reference .get ("sha256" )
104+ == hashlib .sha256 (record_review ).hexdigest ()
105+ )
106+
107+
108+ def _squash_candidate (repo : Path , head_commit : str , source_commit : str ) -> str :
109+ tree = _git (repo , "rev-parse" , f"{ head_commit } )
110+ return subprocess .check_output (
111+ [
112+ "git" ,
113+ "-c" ,
114+ "user.name=GovEngine release gate" ,
115+ "-c" ,
116+ "user.email=release-gate@example.invalid" ,
117+ "commit-tree" ,
118+ tree ,
119+ "-p" ,
120+ source_commit ,
121+ ],
122+ cwd = repo ,
123+ input = "Synthetic exact-squash record candidate\n " ,
124+ text = True ,
125+ ).strip ()
126+
127+
128+ def resolve_release_ab_state (repo : Path ) -> ReleaseABState :
129+ head_commit = _git (repo , "rev-parse" , "--verify" , "HEAD^{commit}" )
130+ review_path = repo / REVIEW_PATH
131+ window_path = repo / WINDOW_PATH
132+ review = _load_json (review_path .read_text (encoding = "utf-8" ))
133+ if not isinstance (review , dict ):
134+ raise ValueError ("rc2 review record must be a JSON object" )
135+
136+ if review == PENDING_REVIEW :
137+ if window_path .exists ():
138+ raise ValueError ("pending rc2 source must not contain an rc2 window" )
139+ return ReleaseABState ("synthetic" , head_commit , None )
140+
141+ if not window_path .exists ():
142+ raise ValueError ("approved rc2 review requires an rc2 window" )
143+ window = _load_json (window_path .read_text (encoding = "utf-8" ))
144+ if not isinstance (window , dict ):
145+ raise ValueError ("rc2 window must be a JSON object" )
146+ source_commit = review .get ("source_commit" )
147+ if (
148+ review .get ("verdict" ) != "approved"
149+ or not isinstance (source_commit , str )
150+ or not FULL_SHA .fullmatch (source_commit )
151+ or window .get ("schema_version" ) != "govengine.rc_window.v2"
152+ or window .get ("version" ) != "1.0.0rc2"
153+ or window .get ("source_commit" ) != source_commit
154+ ):
155+ raise ValueError ("rc2 review and window identity are inconsistent" )
156+ reference = window .get ("security_review" )
157+ current_review = review_path .read_bytes ()
158+ if (
159+ not isinstance (reference , dict )
160+ or reference .get ("path" ) != str (REVIEW_PATH )
161+ or reference .get ("sha256" ) != hashlib .sha256 (current_review ).hexdigest ()
162+ ):
163+ raise ValueError ("rc2 window does not bind the current review record" )
164+ if subprocess .run (
165+ ["git" , "merge-base" , "--is-ancestor" , source_commit , head_commit ],
166+ cwd = repo ,
167+ check = False ,
168+ stdout = subprocess .PIPE ,
169+ stderr = subprocess .PIPE ,
170+ ).returncode != 0 :
171+ raise ValueError ("rc2 source is not an ancestor of the checked commit" )
172+
173+ candidates : list [str ] = []
174+ commits = _git (
175+ repo ,
176+ "rev-list" ,
177+ "--reverse" ,
178+ "--ancestry-path" ,
179+ f"{ source_commit } { head_commit } ,
180+ ).splitlines ()
181+ for commit in commits :
182+ parents = _git (repo , "rev-list" , "--parents" , "-n" , "1" , commit ).split ()[1 :]
183+ if parents != [source_commit ]:
184+ continue
185+ if _matches_authentic_record (repo , commit , source_commit , current_review ):
186+ candidates .append (commit )
187+
188+ expected_changes = [
189+ f"{ status } \t { path } for status , path in EXPECTED_RECORD_CHANGES
190+ ]
191+ aggregate_changes = _git (
192+ repo , "diff" , "--name-status" , source_commit , head_commit
193+ ).splitlines ()
194+ if (
195+ not candidates
196+ and window .get ("status" ) == "prepared"
197+ and aggregate_changes == expected_changes
198+ ):
199+ candidate = _squash_candidate (repo , head_commit , source_commit )
200+ if _matches_authentic_record (repo , candidate , source_commit , current_review ):
201+ candidates .append (candidate )
202+ if len (candidates ) != 1 :
203+ raise ValueError ("exactly one authentic rc2 record child must resolve" )
204+ return ReleaseABState ("authentic" , source_commit , candidates [0 ])
205+
206+
32207def main () -> int :
33208 parser = argparse .ArgumentParser ()
34209 parser .add_argument ("--repo" , type = Path , default = Path ("." ))
35210 parser .add_argument ("--review-commit" , default = "HEAD" )
211+ parser .add_argument ("--resolve-ab-state" , action = "store_true" )
36212 args = parser .parse_args ()
37213 try :
38- print (validate_record_commit (args .repo , args .review_commit ))
39- except (ValueError , subprocess .CalledProcessError ) as error :
214+ if args .resolve_ab_state :
215+ state = resolve_release_ab_state (args .repo )
216+ print (
217+ "\t " .join (
218+ (state .mode , state .source_commit , state .record_commit or "-" )
219+ )
220+ )
221+ else :
222+ print (validate_record_commit (args .repo , args .review_commit ))
223+ except (
224+ OSError ,
225+ ValueError ,
226+ json .JSONDecodeError ,
227+ subprocess .CalledProcessError ,
228+ ) as error :
40229 print (f"release_record_commit_invalid:{ error } )
41230 return 1
42231 return 0
0 commit comments