Coverage for webapp/api/github.py: 65%

1import hmac 

2from hashlib import sha1 

3from os import getenv 

4 

5from webapp import api 

6from webapp.helpers import get_yaml_loader 

7from werkzeug.exceptions import Unauthorized, Forbidden 

8from requests.exceptions import HTTPError 

9 

10import gzip 

11from io import BytesIO 

12import json 

13 

14GITHUB_WEBHOOK_SECRET = getenv("GITHUB_WEBHOOK_SECRET") 

15 

16 

17class InvalidYAML(Exception): 

18 pass 

19 

20 

21class GitHub: 

22 """ 

23 Provides authentication for GitHub users. Helper methods are also provided 

24 for checking organization access and getting user data from the Github API. 

25 """ 

26 

27 REST_API_URL = "https://api.github.com" 

28 GRAPHQL_API_URL = "https://api.github.com/graphql" 

29 RAW_CONTENT_URL = "https://raw.githubusercontent.com" 

30 

31 YAML_LOCATIONS = [ 

32 "snapcraft.yaml", 

33 ".snapcraft.yaml", 

34 "snap/snapcraft.yaml", 

35 "build-aux/snap/snapcraft.yaml", 

36 ] 

37 

38 def __init__(self, access_token=None, session=api.requests.Session()): 

39 self.access_token = access_token 

40 self.session = session 

41 self.session.headers["Accept"] = "application/json" 

42 

43 def _request( 

44 self, method="GET", url="", params={}, data={}, raise_exceptions=True 

45 ): 

46 """ 

47 Makes a raw HTTP request and returns the response. 

48 """ 

49 if self.access_token: 

50 headers = {"Authorization": f"token {self.access_token}"} 

51 else: 

52 headers = {} 

53 

54 response = self.session.request( 

55 method, 

56 f"{self.REST_API_URL}/{url}", 

57 headers=headers, 

58 params=params, 

59 json=data, 

60 ) 

61 

62 if raise_exceptions: 

63 if response.status_code == 401: 

64 raise Unauthorized(response=response) 

65 if response.status_code == 403: 

66 raise Forbidden(response=response) 

67 

68 response.raise_for_status() 

69 

70 return response 

71 

72 def decompress_data(self, data, encoding): 

73 if encoding == "gzip": 

74 with gzip.GzipFile(fileobj=BytesIO(data)) as f: 

75 return f.read().decode( 

76 "utf-8" 

77 ) # Decompress and decode as UTF-8 

78 return data.decode("utf-8") 

79 

80 def get_data_from_response(self, response): 

81 content_encoding = response.headers.get("Content-Encoding", "") 

82 if content_encoding == "gzip": 

83 try: 

84 content = response.content 

85 decompressed_data = self.decompress_data( 

86 content, content_encoding 

87 ) 

88 data = json.loads(decompressed_data) 

89 except Exception: 

90 data = response.json() 

91 else: 

92 data = response.json() 

93 return data 

94 

95 def _gql_request(self, query={}): 

96 """ 

97 Makes a raw HTTP request and returns the response. 

98 """ 

99 if self.access_token: 

100 headers = {"Authorization": f"token {self.access_token}"} 

101 else: 

102 headers = {} 

103 

104 response = self.session.request( 

105 "POST", 

106 self.GRAPHQL_API_URL, 

107 json={"query": query}, 

108 headers=headers, 

109 ) 

110 

111 if response.status_code == 401: 

112 raise Unauthorized(response=response) 

113 if response.status_code == 403: 

114 raise Forbidden 

115 

116 response.raise_for_status() 

117 

118 data = self.get_data_from_response(response) 

119 return data["data"] 

120 

121 def _get_nodes(self, edges): 

122 """ 

123 GraphQL: Return the list of nodes from the edges 

124 """ 

125 return [i["node"] for i in edges] 

126 

127 def get_user(self): 

128 """ 

129 Return some user properties of the current user 

130 """ 

131 gql = """ 

132 { 

133 viewer { 

134 login 

135 name 

136 avatarUrl(size: 100) 

137 } 

138 } 

139 """ 

140 

141 return self._gql_request(gql)["viewer"] 

142 

143 def get_orgs(self, end_cursor=None): 

144 """ 

145 Lists of organizations that the authenticated user has explicit 

146 permission to access. 

147 """ 

148 gql = ( 

149 """ 

150 { 

151 viewer { 

152 organizations(first: 100,""" 

153 + (f'after: "{end_cursor}"' if end_cursor else "") 

154 + """) { 

155 edges { 

156 node { 

157 login 

158 name 

159 } 

160 } 

161 pageInfo { 

162 hasNextPage 

163 endCursor 

164 } 

165 } 

166 } 

167 } 

168 """ 

169 ) 

170 

171 gql_response = self._gql_request(gql)["viewer"]["organizations"] 

172 page_info = gql_response["pageInfo"] 

173 orgs = self._get_nodes(gql_response["edges"]) 

174 

175 if page_info["hasNextPage"]: 

176 next_page = self.get_orgs(page_info["endCursor"]) 

177 orgs.extend(next_page) 

178 

179 return orgs 

180 

181 def get_user_repositories(self, end_cursor=None): 

182 """ 

183 Lists of public repositories from the authenticated user 

184 """ 

185 gql = ( 

186 """{ 

187 viewer { 

188 repositories( 

189 first: 100, 

190 privacy: PUBLIC, 

191 """ 

192 + (f'after: "{end_cursor}"' if end_cursor else "") 

193 + """ 

194 ) { 

195 edges { 

196 node { 

197 name 

198 nameWithOwner 

199 } 

200 } 

201 pageInfo { 

202 hasNextPage 

203 endCursor 

204 } 

205 } 

206 } 

207 }""" 

208 ) 

209 

210 gql_response = self._gql_request(gql)["viewer"]["repositories"] 

211 page_info = gql_response["pageInfo"] 

212 repositories = self._get_nodes(gql_response["edges"]) 

213 

214 if page_info["hasNextPage"]: 

215 next_page = self.get_user_repositories(page_info["endCursor"]) 

216 repositories.extend(next_page) 

217 

218 repos = [ 

219 ( 

220 {**repo, "owner": repo.get("nameWithOwner", "").split("/")[0]} 

221 if "nameWithOwner" in repo and repo.get("nameWithOwner") 

222 else {**repo, "owner": None} 

223 ) 

224 for repo in repositories 

225 ] 

226 

227 return repos 

228 

229 def get_org_repositories(self, org_login, end_cursor=None): 

230 """ 

231 Lists of public repositories from the authenticated user 

232 """ 

233 gql = ( 

234 """{ 

235 viewer { 

236 organization(login: \"""" 

237 + org_login 

238 + """") { 

239 repositories( 

240 first: 100, 

241 privacy: PUBLIC 

242 """ 

243 + (f'after: "{end_cursor}"' if end_cursor else "") 

244 + """ 

245 ) { 

246 edges { 

247 node { 

248 name 

249 nameWithOwner 

250 } 

251 } 

252 pageInfo { 

253 hasNextPage 

254 endCursor 

255 } 

256 } 

257 } 

258 } 

259 }""" 

260 ) 

261 

262 response = self._gql_request(gql)["viewer"]["organization"][ 

263 "repositories" 

264 ] 

265 

266 page_info = response["pageInfo"] 

267 repositories = self._get_nodes(response["edges"]) 

268 

269 if page_info["hasNextPage"]: 

270 next_page = self.get_org_repositories( 

271 org_login, page_info["endCursor"] 

272 ) 

273 repositories.extend(next_page) 

274 

275 repos = [ 

276 ( 

277 {**repo, "owner": repo.get("nameWithOwner", "").split("/")[0]} 

278 if "nameWithOwner" in repo and repo.get("nameWithOwner") 

279 else {**repo, "owner": None} 

280 ) 

281 for repo in repositories 

282 ] 

283 

284 return repos 

285 

286 def check_permissions_over_repo(self, owner, repo, permission="push"): 

287 """ 

288 Return True when the current user has the requested permissions 

289 Possible values: "admin", "push" or "pull" 

290 """ 

291 try: 

292 response = self._request( 

293 "GET", 

294 f"repos/{owner}/{repo}", 

295 raise_exceptions=True, 

296 ) 

297 except Unauthorized: 

298 return False 

299 except Forbidden: 

300 return False 

301 except HTTPError as e: 

302 if e.response.status_code == 404: 

303 return False 

304 

305 data = self.get_data_from_response(response) 

306 response_permissions = data["permissions"] 

307 user_permissions = [ 

308 p for p in response_permissions if response_permissions[p] 

309 ] 

310 

311 return permission in user_permissions 

312 

313 def check_if_repo_exists(self, owner, repo): 

314 """ 

315 Return True if GitHub repo exists 

316 """ 

317 response = self._request( 

318 "GET", 

319 f"repos/{owner}/{repo}", 

320 raise_exceptions=False, 

321 ) 

322 if response.status_code == 404: 

323 return False 

324 elif response.status_code == 200: 

325 return True 

326 elif response.status_code == 401: 

327 raise Unauthorized 

328 elif response.status_code == 403: 

329 raise Forbidden 

330 

331 response.raise_for_status() 

332 

333 def get_snapcraft_yaml_location(self, owner, repo): 

334 """ 

335 Return the snapcraft.yaml file location in the GitHub repo 

336 """ 

337 

338 # It is not possible to use GraphQL without authentication 

339 # for that reason we are doing a call for each location to the REST API 

340 for loc in self.YAML_LOCATIONS: 

341 response = self._request( 

342 "GET", 

343 f"repos/{owner}/{repo}/contents/{loc}", 

344 raise_exceptions=False, 

345 ) 

346 if response.status_code == 404: 

347 continue 

348 elif response.status_code == 200: 

349 return loc 

350 elif response.status_code == 401: 

351 raise Unauthorized 

352 elif response.status_code == 403: 

353 raise Forbidden 

354 

355 response.raise_for_status() 

356 

357 return False 

358 

359 def get_default_branch(self, owner, repo): 

360 response = self._request("GET", f"repos/{owner}/{repo}") 

361 data = self.get_data_from_response(response) 

362 return data["default_branch"] 

363 

364 def get_last_commit(self, owner, repo, branch=None): 

365 if not branch: 

366 branch = self.get_default_branch(owner, repo) 

367 

368 response = self._request( 

369 "GET", f"repos/{owner}/{repo}/commits/{branch}" 

370 ) 

371 data = self.get_data_from_response(response) 

372 return data["sha"] 

373 

374 def get_snapcraft_yaml_data(self, owner, repo, location=None): 

375 """ 

376 Parse the snapcraft.yaml from the repo and return a dict 

377 """ 

378 if not location: 

379 location = self.get_snapcraft_yaml_location(owner, repo) 

380 

381 if location: 

382 # Get last commit to avoid cache issues with raw.github.com 

383 last_commit = self.get_last_commit(owner, repo) 

384 

385 response = self.session.request( 

386 "GET", 

387 f"{self.RAW_CONTENT_URL}/{owner}/{repo}" 

388 f"/{last_commit}/{location}", 

389 ) 

390 

391 yaml = get_yaml_loader() 

392 try: 

393 content_encoding = response.headers.get("Content-Encoding", "") 

394 if content_encoding == "gzip": 

395 try: 

396 content = response.content 

397 data = self.decompress_data(content, content_encoding) 

398 except Exception: 

399 data = response.content 

400 else: 

401 data = response.content 

402 return yaml.load(data) 

403 except Exception: 

404 raise InvalidYAML 

405 

406 return {} 

407 

408 def generate_webhook_secret_for_repo(self, owner, name): 

409 key = bytes(GITHUB_WEBHOOK_SECRET, "UTF-8") 

410 hmac_gen = hmac.new(key, None, sha1) 

411 hmac_gen.update(bytes(owner, "UTF-8")) 

412 hmac_gen.update(bytes(name, "UTF-8")) 

413 return hmac_gen.hexdigest() 

414 

415 def validate_webhook_signature(self, payload, signature): 

416 """ 

417 Generate the payload signature and compare with the given one 

418 """ 

419 key = bytes(GITHUB_WEBHOOK_SECRET, "UTF-8") 

420 hmac_gen = hmac.new(key, payload, sha1) 

421 

422 # Add append prefix to match the GitHub request format 

423 digest = f"sha1={hmac_gen.hexdigest()}" 

424 

425 return hmac.compare_digest(digest, signature) 

426 

427 def validate_bsi_webhook_secret(self, owner, name, payload, signature): 

428 """ 

429 Return True if the webhook contain a valid secret in BSI 

430 """ 

431 secret = self.generate_webhook_secret_for_repo(owner, name) 

432 final_key = bytes(secret, "UTF-8") 

433 final_hmac = hmac.new(final_key, payload, sha1) 

434 

435 # Add append prefix to match the GitHub request format 

436 digest = f"sha1={final_hmac.hexdigest()}" 

437 

438 return hmac.compare_digest(digest, signature) 

439 

440 def get_hooks(self, owner, repo, page=1): 

441 """ 

442 Return all the webhooks in the repo 

443 """ 

444 response = self._request( 

445 "GET", 

446 f"repos/{owner}/{repo}/hooks", 

447 params={"per_page": 100, "page": page}, 

448 ) 

449 hooks = response.json() 

450 

451 if "next" in response.links: 

452 hooks.extend(self.get_hooks(page=page + 1)) 

453 

454 return hooks 

455 

456 def get_hook_by_url(self, owner, repo, url): 

457 """ 

458 Return a webhook from the repo with the url 

459 """ 

460 hooks = self.get_hooks(owner, repo) 

461 

462 for hook in hooks: 

463 if hook["config"]["url"] == url: 

464 return hook 

465 

466 return None 

467 

468 def update_hook_url(self, owner, repo, hook_id, new_url): 

469 """ 

470 Update a webhook to activate it and update the URL 

471 """ 

472 data = { 

473 "active": True, 

474 "config": { 

475 "url": new_url, 

476 "content_type": "json", 

477 "secret": GITHUB_WEBHOOK_SECRET, 

478 }, 

479 } 

480 

481 self._request( 

482 "PATCH", f"repos/{owner}/{repo}/hooks/{hook_id}", data=data 

483 ) 

484 

485 return True 

486 

487 def create_hook(self, owner, repo, hook_url): 

488 """ 

489 Create the webhook in the repo 

490 """ 

491 secret = self.generate_webhook_secret_for_repo(owner, repo) 

492 data = { 

493 "config": { 

494 "url": hook_url, 

495 "content_type": "json", 

496 "secret": secret, 

497 }, 

498 } 

499 

500 self._request("POST", f"repos/{owner}/{repo}/hooks", data=data) 

501 

502 return True 

503 

504 def remove_hook(self, owner, repo, hook_id): 

505 """ 

506 Remove GitHub webhook in a repo 

507 """ 

508 self._request("DELETE", f"repos/{owner}/{repo}/hooks/{hook_id}") 

509 

510 return True