Coverage for extra_codeowners/webhooks.py: 84%

1"""GitHub webhook authentication and trigger extraction."""

3from __future__ import annotations

5import hashlib

6import hmac

7import json

8from dataclasses import dataclass

9from typing import Any, Final

11from extra_codeowners.database import AuthorityRequest, JobRequest

13MAX_WEBHOOK_BYTES: Final = 10 * 1024 * 1024

14PULL_REQUEST_ACTIONS: Final = frozenset(

15 {

16 "opened",

17 "reopened",

18 "synchronize",

19 "ready_for_review",

20 "converted_to_draft",

21 "edited",

22 "labeled",

23 "unlabeled",

24 "review_requested",

25 "review_request_removed",

26 }

27)

28REVIEW_ACTIONS: Final = frozenset({"submitted", "edited", "dismissed"})

29CODEOWNERS_PATHS: Final = frozenset({".github/CODEOWNERS", "CODEOWNERS", "docs/CODEOWNERS"})

32class WebhookError(ValueError):

33 """A webhook cannot be safely authenticated or interpreted."""

36@dataclass(frozen=True, slots=True)

37class VerifiedWebhook:

38 """Authenticated delivery metadata and parsed payload."""

40 delivery_id: str

41 event: str

42 action: str

43 payload: dict[str, Any]

46def verify_webhook(

47 body: bytes,

48 *,

49 signature: str | None,

50 delivery_id: str | None,

51 event: str | None,

52 secret: str,

53) -> VerifiedWebhook:

54 """Authenticate a GitHub webhook before parsing user-controlled JSON."""

55 if len(body) > MAX_WEBHOOK_BYTES:

56 msg = "webhook payload exceeds the 10 MiB limit"

57 raise WebhookError(msg)

58 if not signature or not signature.startswith("sha256="):

59 msg = "missing or malformed X-Hub-Signature-256"

60 raise WebhookError(msg)

61 if not delivery_id or len(delivery_id) > 128:

62 msg = "missing or malformed X-GitHub-Delivery"

63 raise WebhookError(msg)

64 if not event or len(event) > 128:

65 msg = "missing or malformed X-GitHub-Event"

66 raise WebhookError(msg)

67 expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

68 if not hmac.compare_digest(expected, signature):

69 msg = "webhook signature mismatch"

70 raise WebhookError(msg)

71 try:

72 parsed = json.loads(body)

73 except (UnicodeDecodeError, json.JSONDecodeError) as error:

74 msg = "webhook body is not valid JSON"

75 raise WebhookError(msg) from error

76 if not isinstance(parsed, dict):

77 msg = "webhook JSON root must be an object"

78 raise WebhookError(msg)

79 action = parsed.get("action", "")

80 if not isinstance(action, str): 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true

81 msg = "webhook action must be a string"

82 raise WebhookError(msg)

83 return VerifiedWebhook(delivery_id, event, action, parsed)

86def _positive_int(value: Any, field: str) -> int:

87 if not isinstance(value, int) or isinstance(value, bool) or value <= 0: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 msg = f"webhook {field} must be a positive integer"

89 raise WebhookError(msg)

90 return value

93def _repository_name(payload: dict[str, Any]) -> str:

94 repository = payload.get("repository")

95 if not isinstance(repository, dict) or not isinstance(repository.get("full_name"), str): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 msg = "webhook omitted repository.full_name"

97 raise WebhookError(msg)

98 full_name = str(repository["full_name"])

99 if full_name.count("/") != 1 or any(part == "" for part in full_name.split("/")): 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 raise WebhookError("webhook repository.full_name must be owner/repository")

101 return full_name.lower()

102

103

104def _installation_id(payload: dict[str, Any]) -> int:

105 installation = payload.get("installation")

106 if not isinstance(installation, dict): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true

107 msg = "webhook omitted installation"

108 raise WebhookError(msg)

109 return _positive_int(installation.get("id"), "installation.id")

110

111

112def _pull_request_job(webhook: VerifiedWebhook) -> JobRequest:

113 pull = webhook.payload.get("pull_request")

114 if not isinstance(pull, dict): 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true

115 msg = "webhook omitted pull_request"

116 raise WebhookError(msg)

117 number = _positive_int(pull.get("number") or webhook.payload.get("number"), "pull number")

118 head = pull.get("head")

119 head_sha = head.get("sha") if isinstance(head, dict) else None

120 return JobRequest(

121 installation_id=_installation_id(webhook.payload),

122 repository_full_name=_repository_name(webhook.payload),

123 pull_number=number,

124 reason=f"{webhook.event}.{webhook.action or 'received'}",

125 head_sha_hint=head_sha if isinstance(head_sha, str) else None,

126 )

127

128

129def _push_authority_job(

130 webhook: VerifiedWebhook,

131 *,

132 policy_path: str,

133 org_config_repository: str,

134) -> AuthorityRequest | None:

135 payload = webhook.payload

136 if payload.get("deleted") is True: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true

137 return None

138 ref = payload.get("ref")

139 if not isinstance(ref, str) or not ref.startswith("refs/heads/"): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 return None

141 branch = ref.removeprefix("refs/heads/")

142 repository = payload.get("repository")

143 if not isinstance(repository, dict): 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 raise WebhookError("webhook omitted repository")

145 full_name = _repository_name(payload)

146 repository_name = full_name.split("/", 1)[-1]

147

148 if repository_name != org_config_repository.lower():

149 # Every base-branch push can change a pull request's merge base and

150 # therefore its changed files and applicable CODEOWNERS rules. The

151 # fan-out later narrows this event to PRs whose base ref is ``branch``.

152 return AuthorityRequest(

153 installation_id=_installation_id(payload),

154 repository_full_name=full_name,

155 base_ref=branch,

156 reason="push.repository_base",

157 )

158

159 default_branch = repository.get("default_branch")

160 if not isinstance(default_branch, str) or branch != default_branch: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 return None

162

163 # GitHub bounds the commits included in a push payload. Only the shared

164 # organization-policy repository can safely use path filtering, and any

165 # missing, malformed, or truncated evidence must conservatively fan out.

166 commits = payload.get("commits")

167 changed: set[str] = set()

168 malformed = not isinstance(commits, list)

169 if isinstance(commits, list):

170 for commit in commits:

171 if not isinstance(commit, dict): 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 malformed = True

173 continue

174 for field in ("added", "modified", "removed"):

175 paths = commit.get(field)

176 if not isinstance(paths, list) or any(not isinstance(path, str) for path in paths): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 malformed = True

178 continue

179 changed.update(paths)

180 distinct_size = payload.get("distinct_size")

181 if (

182 isinstance(distinct_size, int)

183 and not isinstance(distinct_size, bool)

184 and distinct_size >= 0

185 ):

186 valid_distinct_size = True

187 reported_commit_count = distinct_size

188 else:

189 valid_distinct_size = False

190 reported_commit_count = 0

191 forced = payload.get("forced")

192 truncated = (

193 malformed

194 or not valid_distinct_size

195 or not isinstance(commits, list)

196 or reported_commit_count > len(commits)

197 or forced is not False

198 )

199 if not truncated and policy_path not in changed:

200 return None

201 return AuthorityRequest(

202 installation_id=_installation_id(payload),

203 repository_full_name=None,

204 base_ref=None,

205 reason="push.organization_policy",

206 )

207

208

209def evaluation_job(

210 webhook: VerifiedWebhook,

211 *,

212 policy_path: str = ".github/extra-codeowners.toml",

213 org_config_repository: str = ".github",

214) -> JobRequest | AuthorityRequest | None:

215 """Map a verified delivery to evaluation or authority fan-out work."""

216 if webhook.event == "pull_request":

217 return _pull_request_job(webhook) if webhook.action in PULL_REQUEST_ACTIONS else None

218 if webhook.event == "pull_request_review":

219 return _pull_request_job(webhook) if webhook.action in REVIEW_ACTIONS else None

220 if webhook.event == "check_run" and webhook.action == "rerequested":

221 check_run = webhook.payload.get("check_run")

222 if not isinstance(check_run, dict): 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true

223 raise WebhookError("webhook omitted check_run")

224 pulls = check_run.get("pull_requests")

225 if not isinstance(pulls, list) or len(pulls) != 1 or not isinstance(pulls[0], dict): 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true

226 return None

227 number = _positive_int(pulls[0].get("number"), "check_run pull number")

228 head_sha = check_run.get("head_sha")

229 return JobRequest(

230 installation_id=_installation_id(webhook.payload),

231 repository_full_name=_repository_name(webhook.payload),

232 pull_number=number,

233 reason="check_run.rerequested",

234 head_sha_hint=head_sha if isinstance(head_sha, str) else None,

235 )

236 if webhook.event == "push":

237 return _push_authority_job(

238 webhook,

239 policy_path=policy_path,

240 org_config_repository=org_config_repository,

241 )

242 if webhook.event == "repository":

243 repository = webhook.payload.get("repository")

244 if not isinstance(repository, dict): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 raise WebhookError("webhook omitted repository")

246 full_name = _repository_name(webhook.payload)

247 repository_name = full_name.split("/", 1)[-1]

248 changes = webhook.payload.get("changes")

249 previous_name: str | None = None

250 default_branch_changed = False

251 if isinstance(changes, dict): 251 ↛ 258line 251 didn't jump to line 258 because the condition on line 251 was always true

252 repository_change = changes.get("repository")

253 if isinstance(repository_change, dict):

254 name_change = repository_change.get("name")

255 if isinstance(name_change, dict) and isinstance(name_change.get("from"), str): 255 ↛ 257line 255 didn't jump to line 257 because the condition on line 255 was always true

256 previous_name = str(name_change["from"])

257 default_branch_changed = "default_branch" in changes

258 org_repository_affected = repository_name == org_config_repository.lower() or (

259 previous_name is not None and previous_name.lower() == org_config_repository.lower()

260 )

261 if org_repository_affected and (

262 webhook.action in {"deleted", "renamed", "transferred"}

263 or (webhook.action == "edited" and default_branch_changed)

264 ):

265 return AuthorityRequest(

266 installation_id=_installation_id(webhook.payload),

267 repository_full_name=None,

268 base_ref=None,

269 reason=f"repository.{webhook.action}",

270 )

271 if webhook.action in {"renamed", "transferred", "unarchived"}:

272 # Mutable repository names are API routes, not durable identities.

273 # Fence the whole installation so an old-name in-flight evaluation

274 # cannot race work discovered under the new route. Unarchive also

275 # revisits checks that were intentionally skipped while archived.

276 return AuthorityRequest(

277 installation_id=_installation_id(webhook.payload),

278 repository_full_name=None,

279 base_ref=None,

280 reason=f"repository.{webhook.action}",

281 )

282 return None

283 if webhook.event in {"label", "member", "team_add"}:

284 return AuthorityRequest(

285 installation_id=_installation_id(webhook.payload),

286 repository_full_name=_repository_name(webhook.payload),

287 base_ref=None,

288 reason=f"{webhook.event}.{webhook.action or 'received'}",

289 )

290 if webhook.event in {"membership", "organization", "team"}:

291 return AuthorityRequest(

292 installation_id=_installation_id(webhook.payload),

293 repository_full_name=None,

294 base_ref=None,

295 reason=f"{webhook.event}.{webhook.action or 'received'}",

296 )

297 if webhook.event == "installation" and webhook.action in {

298 "created",

299 "unsuspend",

300 "new_permissions_accepted",

301 }:

302 return AuthorityRequest(

303 installation_id=_installation_id(webhook.payload),

304 repository_full_name=None,

305 base_ref=None,

306 reason=f"installation.{webhook.action}",

307 )

308 if webhook.event == "installation_repositories":

309 if webhook.action == "added": 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true

310 return AuthorityRequest(

311 installation_id=_installation_id(webhook.payload),

312 repository_full_name=None,

313 base_ref=None,

314 reason="installation_repositories.added",

315 )

316 if webhook.action == "removed": 316 ↛ 351line 316 didn't jump to line 351 because the condition on line 316 was always true

317 removed = webhook.payload.get("repositories_removed")

318 malformed = not isinstance(removed, list) or not removed

319 organization_policy_removed = False

320 if isinstance(removed, list):

321 for repository in removed:

322 if not isinstance(repository, dict): 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true

323 malformed = True

324 continue

325 removed_full_name = repository.get("full_name")

326 if (

327 not isinstance(removed_full_name, str)

328 or removed_full_name.count("/") != 1

329 or any(part == "" for part in removed_full_name.split("/"))

330 ):

331 malformed = True

332 continue

333 repository_name = removed_full_name.rsplit("/", 1)[-1]

334 organization_policy_removed |= (

335 repository_name.lower() == org_config_repository.lower()

336 )

337 if malformed or organization_policy_removed:

338 # Losing the shared policy source can invalidate application

339 # enrollment across every still-accessible target repository.

340 # Malformed evidence is treated as that security-sensitive case.

341 return AuthorityRequest(

342 installation_id=_installation_id(webhook.payload),

343 repository_full_name=None,

344 base_ref=None,

345 reason="installation_repositories.removed",

346 )

347 # The App has already lost the capability needed to revoke a check

348 # in an ordinary removed target repository. Operators must follow

349 # the documented access-removal sequence for that case.

350 return None

351 if webhook.event == "installation_target" and webhook.action == "renamed": 351 ↛ 358line 351 didn't jump to line 358 because the condition on line 351 was always true

352 return AuthorityRequest(

353 installation_id=_installation_id(webhook.payload),

354 repository_full_name=None,

355 base_ref=None,

356 reason="installation_target.renamed",

357 )

358 return None