Coverage for webapp/api/github.py: 66%
214 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-20 22:09 +0000
1import hmac
2from hashlib import sha1
3from os import getenv
5from webapp import api
6from webapp.helpers import get_yaml_loader
7from werkzeug.exceptions import Unauthorized, Forbidden
8from requests.exceptions import HTTPError
10import gzip
11from io import BytesIO
12import json
14GITHUB_WEBHOOK_SECRET = getenv("GITHUB_WEBHOOK_SECRET")
16REPOSITORY_CHECK_TIMEOUT = 5
19def repository_is_public(repository, session=None):
20 """Whether ``owner/repo`` can still be read on github.com.
22 A deleted, renamed or private repository leaves the recipe pointing at a
23 URL that 404s, so a provenance link built from it would invite the user
24 to verify a commit they cannot open. Uses HEAD against the web UI, not
25 the REST API, which caps anonymous callers at 60 requests/hour.
27 Fails open: only a definitive 404 counts as gone, so a GitHub wobble
28 cannot erase provenance we legitimately hold.
29 """
30 if not repository:
31 return True
33 try:
34 response = (session or api.requests.Session()).head(
35 f"https://github.com/{repository}",
36 timeout=REPOSITORY_CHECK_TIMEOUT,
37 allow_redirects=True,
38 )
39 except Exception:
40 return True
42 return response.status_code != 404
45class InvalidYAML(Exception):
46 pass
49class GitHub:
50 """
51 Provides authentication for GitHub users. Helper methods are also provided
52 for checking organization access and getting user data from the Github API.
53 """
55 REST_API_URL = "https://api.github.com"
56 GRAPHQL_API_URL = "https://api.github.com/graphql"
57 RAW_CONTENT_URL = "https://raw.githubusercontent.com"
59 YAML_LOCATIONS = [
60 "snapcraft.yaml",
61 ".snapcraft.yaml",
62 "snap/snapcraft.yaml",
63 "build-aux/snap/snapcraft.yaml",
64 ]
66 def __init__(self, access_token=None, session=api.requests.Session()):
67 self.access_token = access_token
68 self.session = session
69 self.session.headers["Accept"] = "application/json"
71 def _request(
72 self, method="GET", url="", params={}, data={}, raise_exceptions=True
73 ):
74 """
75 Makes a raw HTTP request and returns the response.
76 """
77 if self.access_token:
78 headers = {"Authorization": f"token {self.access_token}"}
79 else:
80 headers = {}
82 response = self.session.request(
83 method,
84 f"{self.REST_API_URL}/{url}",
85 headers=headers,
86 params=params,
87 json=data,
88 )
90 if raise_exceptions:
91 if response.status_code == 401:
92 raise Unauthorized(response=response)
93 if response.status_code == 403:
94 raise Forbidden(response=response)
96 response.raise_for_status()
98 return response
100 def decompress_data(self, data, encoding):
101 if encoding == "gzip":
102 with gzip.GzipFile(fileobj=BytesIO(data)) as f:
103 return f.read().decode(
104 "utf-8"
105 ) # Decompress and decode as UTF-8
106 return data.decode("utf-8")
108 def get_data_from_response(self, response):
109 content_encoding = response.headers.get("Content-Encoding", "")
110 if content_encoding == "gzip":
111 try:
112 content = response.content
113 decompressed_data = self.decompress_data(
114 content, content_encoding
115 )
116 data = json.loads(decompressed_data)
117 except Exception:
118 data = response.json()
119 else:
120 data = response.json()
121 return data
123 def _gql_request(self, query={}):
124 """
125 Makes a raw HTTP request and returns the response.
126 """
127 if self.access_token:
128 headers = {"Authorization": f"token {self.access_token}"}
129 else:
130 headers = {}
132 response = self.session.request(
133 "POST",
134 self.GRAPHQL_API_URL,
135 json={"query": query},
136 headers=headers,
137 )
139 if response.status_code == 401:
140 raise Unauthorized(response=response)
141 if response.status_code == 403:
142 raise Forbidden
144 response.raise_for_status()
146 data = self.get_data_from_response(response)
147 return data["data"]
149 def _get_nodes(self, edges):
150 """
151 GraphQL: Return the list of nodes from the edges
152 """
153 return [i["node"] for i in edges]
155 def get_user(self):
156 """
157 Return some user properties of the current user
158 """
159 gql = """
160 {
161 viewer {
162 login
163 name
164 avatarUrl(size: 100)
165 }
166 }
167 """
169 return self._gql_request(gql)["viewer"]
171 def get_orgs(self, end_cursor=None):
172 """
173 Lists of organizations that the authenticated user has explicit
174 permission to access.
175 """
176 gql = (
177 """
178 {
179 viewer {
180 organizations(first: 100,"""
181 + (f'after: "{end_cursor}"' if end_cursor else "")
182 + """) {
183 edges {
184 node {
185 login
186 name
187 }
188 }
189 pageInfo {
190 hasNextPage
191 endCursor
192 }
193 }
194 }
195 }
196 """
197 )
199 gql_response = self._gql_request(gql)["viewer"]["organizations"]
200 page_info = gql_response["pageInfo"]
201 orgs = self._get_nodes(gql_response["edges"])
203 if page_info["hasNextPage"]:
204 next_page = self.get_orgs(page_info["endCursor"])
205 orgs.extend(next_page)
207 return orgs
209 def get_user_repositories(self, end_cursor=None):
210 """
211 Lists of public repositories from the authenticated user
212 """
213 gql = (
214 """{
215 viewer {
216 repositories(
217 first: 100,
218 privacy: PUBLIC,
219 """
220 + (f'after: "{end_cursor}"' if end_cursor else "")
221 + """
222 ) {
223 edges {
224 node {
225 name
226 nameWithOwner
227 }
228 }
229 pageInfo {
230 hasNextPage
231 endCursor
232 }
233 }
234 }
235 }"""
236 )
238 gql_response = self._gql_request(gql)["viewer"]["repositories"]
239 page_info = gql_response["pageInfo"]
240 repositories = self._get_nodes(gql_response["edges"])
242 if page_info["hasNextPage"]:
243 next_page = self.get_user_repositories(page_info["endCursor"])
244 repositories.extend(next_page)
246 repos = [
247 (
248 {**repo, "owner": repo.get("nameWithOwner", "").split("/")[0]}
249 if "nameWithOwner" in repo and repo.get("nameWithOwner")
250 else {**repo, "owner": None}
251 )
252 for repo in repositories
253 ]
255 return repos
257 def get_org_repositories(self, org_login, end_cursor=None):
258 """
259 Lists of public repositories from the authenticated user
260 """
261 gql = (
262 """{
263 viewer {
264 organization(login: \""""
265 + org_login
266 + """") {
267 repositories(
268 first: 100,
269 privacy: PUBLIC
270 """
271 + (f'after: "{end_cursor}"' if end_cursor else "")
272 + """
273 ) {
274 edges {
275 node {
276 name
277 nameWithOwner
278 }
279 }
280 pageInfo {
281 hasNextPage
282 endCursor
283 }
284 }
285 }
286 }
287 }"""
288 )
290 response = self._gql_request(gql)["viewer"]["organization"][
291 "repositories"
292 ]
294 page_info = response["pageInfo"]
295 repositories = self._get_nodes(response["edges"])
297 if page_info["hasNextPage"]:
298 next_page = self.get_org_repositories(
299 org_login, page_info["endCursor"]
300 )
301 repositories.extend(next_page)
303 repos = [
304 (
305 {**repo, "owner": repo.get("nameWithOwner", "").split("/")[0]}
306 if "nameWithOwner" in repo and repo.get("nameWithOwner")
307 else {**repo, "owner": None}
308 )
309 for repo in repositories
310 ]
312 return repos
314 def check_permissions_over_repo(self, owner, repo, permission="push"):
315 """
316 Return True when the current user has the requested permissions
317 Possible values: "admin", "push" or "pull"
318 """
319 try:
320 response = self._request(
321 "GET",
322 f"repos/{owner}/{repo}",
323 raise_exceptions=True,
324 )
325 except Unauthorized:
326 return False
327 except Forbidden:
328 return False
329 except HTTPError as e:
330 if e.response.status_code == 404:
331 return False
333 data = self.get_data_from_response(response)
334 response_permissions = data["permissions"]
335 user_permissions = [
336 p for p in response_permissions if response_permissions[p]
337 ]
339 return permission in user_permissions
341 def check_if_repo_exists(self, owner, repo):
342 """
343 Return True if GitHub repo exists
344 """
345 response = self._request(
346 "GET",
347 f"repos/{owner}/{repo}",
348 raise_exceptions=False,
349 )
350 if response.status_code == 404:
351 return False
352 elif response.status_code == 200:
353 return True
354 elif response.status_code == 401:
355 raise Unauthorized
356 elif response.status_code == 403:
357 raise Forbidden
359 response.raise_for_status()
361 def get_snapcraft_yaml_location(self, owner, repo):
362 """
363 Return the snapcraft.yaml file location in the GitHub repo
364 """
366 # It is not possible to use GraphQL without authentication
367 # for that reason we are doing a call for each location to the REST API
368 for loc in self.YAML_LOCATIONS:
369 response = self._request(
370 "GET",
371 f"repos/{owner}/{repo}/contents/{loc}",
372 raise_exceptions=False,
373 )
374 if response.status_code == 404:
375 continue
376 elif response.status_code == 200:
377 return loc
378 elif response.status_code == 401:
379 raise Unauthorized
380 elif response.status_code == 403:
381 raise Forbidden
383 response.raise_for_status()
385 return False
387 def get_default_branch(self, owner, repo):
388 response = self._request("GET", f"repos/{owner}/{repo}")
389 data = self.get_data_from_response(response)
390 return data["default_branch"]
392 def get_last_commit(self, owner, repo, branch=None):
393 if not branch:
394 branch = self.get_default_branch(owner, repo)
396 response = self._request(
397 "GET", f"repos/{owner}/{repo}/commits/{branch}"
398 )
399 data = self.get_data_from_response(response)
400 return data["sha"]
402 def get_snapcraft_yaml_data(self, owner, repo, location=None):
403 """
404 Parse the snapcraft.yaml from the repo and return a dict
405 """
406 if not location:
407 location = self.get_snapcraft_yaml_location(owner, repo)
409 if location:
410 # Get last commit to avoid cache issues with raw.github.com
411 last_commit = self.get_last_commit(owner, repo)
413 response = self.session.request(
414 "GET",
415 f"{self.RAW_CONTENT_URL}/{owner}/{repo}"
416 f"/{last_commit}/{location}",
417 )
419 yaml = get_yaml_loader()
420 try:
421 content_encoding = response.headers.get("Content-Encoding", "")
422 if content_encoding == "gzip":
423 try:
424 content = response.content
425 data = self.decompress_data(content, content_encoding)
426 except Exception:
427 data = response.content
428 else:
429 data = response.content
430 return yaml.load(data)
431 except Exception:
432 raise InvalidYAML
434 return {}
436 def generate_webhook_secret_for_repo(self, owner, name):
437 key = bytes(GITHUB_WEBHOOK_SECRET, "UTF-8")
438 hmac_gen = hmac.new(key, None, sha1)
439 hmac_gen.update(bytes(owner, "UTF-8"))
440 hmac_gen.update(bytes(name, "UTF-8"))
441 return hmac_gen.hexdigest()
443 def validate_webhook_signature(self, payload, signature):
444 """
445 Generate the payload signature and compare with the given one
446 """
447 key = bytes(GITHUB_WEBHOOK_SECRET, "UTF-8")
448 hmac_gen = hmac.new(key, payload, sha1)
450 # Add append prefix to match the GitHub request format
451 digest = f"sha1={hmac_gen.hexdigest()}"
453 return hmac.compare_digest(digest, signature)
455 def validate_bsi_webhook_secret(self, owner, name, payload, signature):
456 """
457 Return True if the webhook contain a valid secret in BSI
458 """
459 secret = self.generate_webhook_secret_for_repo(owner, name)
460 final_key = bytes(secret, "UTF-8")
461 final_hmac = hmac.new(final_key, payload, sha1)
463 # Add append prefix to match the GitHub request format
464 digest = f"sha1={final_hmac.hexdigest()}"
466 return hmac.compare_digest(digest, signature)
468 def get_hooks(self, owner, repo, page=1):
469 """
470 Return all the webhooks in the repo
471 """
472 response = self._request(
473 "GET",
474 f"repos/{owner}/{repo}/hooks",
475 params={"per_page": 100, "page": page},
476 )
477 hooks = response.json()
479 if "next" in response.links:
480 hooks.extend(self.get_hooks(page=page + 1))
482 return hooks
484 def get_hook_by_url(self, owner, repo, url):
485 """
486 Return a webhook from the repo with the url
487 """
488 hooks = self.get_hooks(owner, repo)
490 for hook in hooks:
491 if hook["config"]["url"] == url:
492 return hook
494 return None
496 def update_hook_url(self, owner, repo, hook_id, new_url):
497 """
498 Update a webhook to activate it and update the URL
499 """
500 data = {
501 "active": True,
502 "config": {
503 "url": new_url,
504 "content_type": "json",
505 "secret": GITHUB_WEBHOOK_SECRET,
506 },
507 }
509 self._request(
510 "PATCH", f"repos/{owner}/{repo}/hooks/{hook_id}", data=data
511 )
513 return True
515 def create_hook(self, owner, repo, hook_url):
516 """
517 Create the webhook in the repo
518 """
519 secret = self.generate_webhook_secret_for_repo(owner, repo)
520 data = {
521 "config": {
522 "url": hook_url,
523 "content_type": "json",
524 "secret": secret,
525 },
526 }
528 self._request("POST", f"repos/{owner}/{repo}/hooks", data=data)
530 return True
532 def remove_hook(self, owner, repo, hook_id):
533 """
534 Remove GitHub webhook in a repo
535 """
536 self._request("DELETE", f"repos/{owner}/{repo}/hooks/{hook_id}")
538 return True