Coverage for webapp/publisher/snaps/build_views.py: 16%
253 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-05 22:06 +0000
1# Standard library
2import os
3from hashlib import md5
5# Packages
6import flask
7from canonicalwebteam.store_api.dashboard import Dashboard
9from requests.exceptions import HTTPError
11# Local
12from webapp.helpers import api_publisher_session, launchpad
13from webapp.api.github import GitHub, InvalidYAML
14from webapp.decorators import login_required
15from webapp.extensions import csrf
16from webapp.publisher.snaps.builds import map_build_and_upload_states
17from werkzeug.exceptions import Unauthorized
19GITHUB_SNAPCRAFT_USER_TOKEN = os.getenv("GITHUB_SNAPCRAFT_USER_TOKEN")
20GITHUB_WEBHOOK_HOST_URL = os.getenv("GITHUB_WEBHOOK_HOST_URL")
21BUILDS_PER_PAGE = 15
22dashboard = Dashboard(api_publisher_session)
25def get_builds(lp_snap, selection):
26 builds = launchpad.get_snap_builds(lp_snap["store_name"])
28 total_builds = len(builds)
30 builds = builds[selection]
32 snap_builds = []
33 builders_status = None
35 for build in builds:
36 status = map_build_and_upload_states(
37 build["buildstate"], build["store_upload_status"]
38 )
40 snap_build = {
41 "id": build["self_link"].split("/")[-1],
42 "arch_tag": build["arch_tag"],
43 "datebuilt": build["datebuilt"],
44 "duration": build["duration"],
45 "logs": build["build_log_url"],
46 "revision_id": build["revision_id"],
47 "status": status,
48 "title": build["title"],
49 "queue_time": None,
50 }
52 if build["buildstate"] == "Needs building":
53 if not builders_status:
54 builders_status = launchpad.get_builders_status()
56 snap_build["queue_time"] = builders_status[build["arch_tag"]][
57 "estimated_duration"
58 ]
60 snap_builds.append(snap_build)
62 return {
63 "total_builds": total_builds,
64 "snap_builds": snap_builds,
65 }
68@login_required
69def get_snap_repo(snap_name):
70 res = {"message": "", "success": True}
71 data = {"github_orgs": [], "github_repository": None, "github_user": None}
73 details = dashboard.get_snap_info(flask.session, snap_name)
75 # API call to make users without needed permissions refresh the session
76 # Users needs package_upload_request permission to use this feature
77 dashboard.get_package_upload_macaroon(
78 session=flask.session, snap_name=snap_name, channels=["edge"]
79 )
81 # Get built snap in launchpad with this store name
82 lp_snap = launchpad.get_snap_by_store_name(details["snap_name"])
84 if lp_snap:
85 # In this case we can use the GitHub user account or
86 # the Snapcraft GitHub user to check the snapcraft.yaml
87 github = GitHub(
88 flask.session.get(
89 "github_auth_secret", GITHUB_SNAPCRAFT_USER_TOKEN
90 )
91 )
93 # Git repository without GitHub hostname
94 data["github_repository"] = lp_snap["git_repository_url"][19:]
95 github_owner, github_repo = data["github_repository"].split("/")
97 if not github.check_if_repo_exists(github_owner, github_repo):
98 data["success"] = False
99 data["message"] = "This app has been revoked"
101 if github.get_user():
102 data["github_user"] = github.get_user()
103 data["github_orgs"] = github.get_orgs()
105 else:
106 data["github_repository"] = None
107 github = GitHub(flask.session.get("github_auth_secret"))
109 if github.get_user():
110 data["github_user"] = github.get_user()
111 data["github_orgs"] = github.get_orgs()
112 else:
113 data["success"] = False
114 data["message"] = "Unauthorized"
116 res["data"] = data
118 return flask.jsonify(res)
121@login_required
122def get_snap_builds_page(snap_name):
123 # If this fails, the page will 404
124 dashboard.get_snap_info(flask.session, snap_name)
125 return flask.render_template("store/publisher.html", snap_name=snap_name)
128@login_required
129def get_snap_build_page(snap_name, build_id):
130 # If this fails, the page will 404
131 dashboard.get_snap_info(flask.session, snap_name)
132 return flask.render_template(
133 "store/publisher.html", snap_name=snap_name, build_id=build_id
134 )
137@login_required
138def get_snap_builds(snap_name):
139 res = {"message": "", "success": True}
140 data = {"snap_builds": [], "total_builds": 0}
142 details = dashboard.get_snap_info(flask.session, snap_name)
143 start = flask.request.args.get("start", 0, type=int)
144 size = flask.request.args.get("size", 15, type=int)
145 build_slice = slice(start, size)
147 # Get built snap in launchpad with this store name
148 lp_snap = launchpad.get_snap_by_store_name(details["snap_name"])
150 if lp_snap:
151 data.update(get_builds(lp_snap, build_slice))
153 res["data"] = data
155 return flask.jsonify(res)
158@login_required
159def get_snap_build(snap_name, build_id):
160 details = dashboard.get_snap_info(flask.session, snap_name)
162 context = {
163 "snap_id": details["snap_id"],
164 "snap_name": details["snap_name"],
165 "snap_title": details["title"],
166 "snap_build": {},
167 }
169 # Get build by snap name and build_id
170 lp_build = launchpad.get_snap_build(details["snap_name"], build_id)
172 if lp_build:
173 status = map_build_and_upload_states(
174 lp_build["buildstate"], lp_build["store_upload_status"]
175 )
176 context["snap_build"] = {
177 "id": lp_build["self_link"].split("/")[-1],
178 "arch_tag": lp_build["arch_tag"],
179 "datebuilt": lp_build["datebuilt"],
180 "duration": lp_build["duration"],
181 "logs": lp_build["build_log_url"],
182 "revision_id": lp_build["revision_id"],
183 "status": status,
184 "title": lp_build["title"],
185 }
187 if context["snap_build"]["logs"]:
188 context["raw_logs"] = launchpad.get_snap_build_log(
189 details["snap_name"], build_id
190 )
192 return flask.jsonify({"data": context, "success": True})
195def validate_repo(github_token, snap_name, gh_owner, gh_repo):
196 github = GitHub(github_token)
197 result = {"success": True}
198 yaml_location = github.get_snapcraft_yaml_location(gh_owner, gh_repo)
200 # The snapcraft.yaml is not present
201 if not yaml_location:
202 result["success"] = False
203 result["error"] = {
204 "type": "MISSING_YAML_FILE",
205 "message": (
206 "Missing snapcraft.yaml: this repo needs a snapcraft.yaml "
207 "file, so that Snapcraft can make it buildable, installable "
208 "and runnable."
209 ),
210 }
211 # The property name inside the yaml file doesn't match the snap
212 else:
213 try:
214 gh_snap_name = github.get_snapcraft_yaml_data(
215 gh_owner, gh_repo
216 ).get("name")
218 if gh_snap_name != snap_name:
219 result["success"] = False
220 result["error"] = {
221 "type": "SNAP_NAME_DOES_NOT_MATCH",
222 "message": (
223 "Name mismatch: the snapcraft.yaml uses the snap "
224 f'name "{gh_snap_name}", but you\'ve registered'
225 f' the name "{snap_name}". Update your '
226 "snapcraft.yaml to continue."
227 ),
228 "yaml_location": yaml_location,
229 "gh_snap_name": gh_snap_name,
230 }
231 except InvalidYAML:
232 result["success"] = False
233 result["error"] = {
234 "type": "INVALID_YAML_FILE",
235 "message": (
236 "Invalid snapcraft.yaml: there was an issue parsing the "
237 f"snapcraft.yaml for {snap_name}."
238 ),
239 }
241 return result
244@login_required
245def get_validate_repo(snap_name):
246 details = dashboard.get_snap_info(flask.session, snap_name)
248 owner, repo = flask.request.args.get("repo").split("/")
250 return flask.jsonify(
251 validate_repo(
252 flask.session.get("github_auth_secret"),
253 details["snap_name"],
254 owner,
255 repo,
256 )
257 )
260@login_required
261def post_snap_builds(snap_name):
262 details = dashboard.get_snap_info(flask.session, snap_name)
264 # Don't allow changes from Admins that are no contributors
265 account_snaps = dashboard.get_account_snaps(flask.session)
267 if snap_name not in account_snaps:
268 flask.flash(
269 "You do not have permissions to modify this Snap", "negative"
270 )
271 return flask.redirect(
272 flask.url_for(".get_snap_builds", snap_name=snap_name)
273 )
275 redirect_url = flask.url_for(".get_snap_builds", snap_name=snap_name)
277 # Get built snap in launchpad with this store name
278 github = GitHub(flask.session.get("github_auth_secret"))
279 owner, repo = flask.request.form.get("github_repository").split("/")
281 if not github.check_permissions_over_repo(owner, repo):
282 flask.flash(
283 "The repository doesn't exist or you don't have"
284 " enough permissions",
285 "negative",
286 )
287 return flask.redirect(redirect_url)
289 repo_validation = validate_repo(
290 flask.session.get("github_auth_secret"), snap_name, owner, repo
291 )
293 if not repo_validation["success"]:
294 flask.flash(repo_validation["error"]["message"], "negative")
295 return flask.redirect(redirect_url)
297 lp_snap = launchpad.get_snap_by_store_name(details["snap_name"])
298 git_url = f"https://github.com/{owner}/{repo}"
300 if not lp_snap:
301 lp_snap_name = md5(git_url.encode("UTF-8")).hexdigest()
303 try:
304 repo_exist = launchpad.get_snap(lp_snap_name)
305 except HTTPError as e:
306 if e.response.status_code == 404:
307 repo_exist = False
308 else:
309 raise e
311 if repo_exist:
312 flask.flash(
313 "The specified repository is being used by another snap:"
314 f" {repo_exist['store_name']}",
315 "negative",
316 )
317 return flask.redirect(redirect_url)
319 macaroon = dashboard.get_package_upload_macaroon(
320 session=flask.session, snap_name=snap_name, channels=["edge"]
321 )["macaroon"]
323 launchpad.create_snap(snap_name, git_url, macaroon)
325 flask.flash(
326 "The GitHub repository was linked successfully.", "positive"
327 )
329 # Create webhook in the repo, it should also trigger the first build
330 github_hook_url = (
331 f"{GITHUB_WEBHOOK_HOST_URL}api/{snap_name}/webhook/notify"
332 )
333 try:
334 hook = github.get_hook_by_url(owner, repo, github_hook_url)
336 # We create the webhook if doesn't exist already in this repo
337 if not hook:
338 github.create_hook(owner, repo, github_hook_url)
339 except HTTPError:
340 flask.flash(
341 "The GitHub Webhook could not be created. "
342 "Please trigger a new build manually.",
343 "caution",
344 )
346 elif lp_snap["git_repository_url"] != git_url:
347 # In the future, create a new record, delete the old one
348 raise AttributeError(
349 f"Snap {snap_name} already has a build repository associated"
350 )
352 return flask.redirect(redirect_url)
355@login_required
356def post_build(snap_name):
357 # Don't allow builds from no contributors
358 account_snaps = dashboard.get_account_snaps(flask.session)
360 if snap_name not in account_snaps:
361 return flask.jsonify(
362 {
363 "success": False,
364 "error": {
365 "type": "FORBIDDEN",
366 "message": "You are not allowed to request "
367 "builds for this snap",
368 },
369 }
370 )
372 try:
373 if launchpad.is_snap_building(snap_name):
374 launchpad.cancel_snap_builds(snap_name)
376 build_id = launchpad.build_snap(snap_name)
378 except HTTPError as e:
379 return flask.jsonify(
380 {
381 "success": False,
382 "error": {
383 "message": "An error happened building "
384 "this snap, please try again."
385 },
386 "details": e.response.text,
387 "status_code": e.response.status_code,
388 }
389 )
391 return flask.jsonify({"success": True, "build_id": build_id})
394@login_required
395def check_build_request(snap_name, build_id):
396 # Don't allow builds from no contributors
397 account_snaps = dashboard.get_account_snaps(flask.session)
399 if snap_name not in account_snaps:
400 return flask.jsonify(
401 {
402 "success": False,
403 "error": {
404 "type": "FORBIDDEN",
405 "message": "You are not allowed to request "
406 "builds for this snap",
407 },
408 }
409 )
411 try:
412 response = launchpad.get_snap_build_request(snap_name, build_id)
413 except HTTPError as e:
414 # Timeout or not found from Launchpad
415 if e.response.status_code in [408, 404]:
416 return flask.jsonify(
417 {
418 "success": False,
419 "error": {
420 "message": "An error happened building "
421 "this snap, please try again."
422 },
423 }
424 )
425 raise e
427 error_message = None
428 if response["error_message"]:
429 error_message = response["error_message"].split(" HEAD:")[0]
431 return flask.jsonify(
432 {
433 "success": True,
434 "status": response["status"],
435 "error": {"message": error_message},
436 }
437 )
440@login_required
441def post_disconnect_repo(snap_name):
442 details = dashboard.get_snap_info(flask.session, snap_name)
444 lp_snap = launchpad.get_snap_by_store_name(snap_name)
445 launchpad.delete_snap(details["snap_name"])
447 # Try to remove the GitHub webhook if possible
448 if flask.session.get("github_auth_secret"):
449 github = GitHub(flask.session.get("github_auth_secret"))
451 try:
452 gh_owner, gh_repo = lp_snap["git_repository_url"][19:].split("/")
454 old_hook = github.get_hook_by_url(
455 gh_owner,
456 gh_repo,
457 f"{GITHUB_WEBHOOK_HOST_URL}api/{snap_name}/webhook/notify",
458 )
460 if old_hook:
461 github.remove_hook(
462 gh_owner,
463 gh_repo,
464 old_hook["id"],
465 )
466 except HTTPError:
467 pass
469 return flask.redirect(
470 flask.url_for(".get_snap_builds", snap_name=snap_name)
471 )
474@csrf.exempt
475def post_github_webhook(snap_name=None, github_owner=None, github_repo=None):
476 payload = flask.request.json
477 repo_url = payload["repository"]["html_url"]
478 gh_owner = payload["repository"]["owner"]["login"]
479 gh_repo = payload["repository"]["name"]
480 gh_default_branch = payload["repository"]["default_branch"]
482 # The first payload after the webhook creation
483 # doesn't contain a "ref" key
484 if "ref" in payload:
485 gh_event_branch = payload["ref"][11:]
486 else:
487 gh_event_branch = gh_default_branch
489 # Check the push event is in the default branch
490 if gh_default_branch != gh_event_branch:
491 return ("The push event is not for the default branch", 200)
493 if snap_name:
494 lp_snap = launchpad.get_snap_by_store_name(snap_name)
495 else:
496 lp_snap = launchpad.get_snap(md5(repo_url.encode("UTF-8")).hexdigest())
498 if not lp_snap:
499 return ("This repository is not linked with any Snap", 403)
501 # Check that this is the repo for this snap
502 if lp_snap["git_repository_url"] != repo_url:
503 return ("The repository does not match the one used by this Snap", 403)
505 github = GitHub()
507 signature = flask.request.headers.get("X-Hub-Signature")
509 if not github.validate_webhook_signature(flask.request.data, signature):
510 if not github.validate_bsi_webhook_secret(
511 gh_owner, gh_repo, flask.request.data, signature
512 ):
513 return ("Invalid secret", 403)
515 validation = validate_repo(
516 GITHUB_SNAPCRAFT_USER_TOKEN, lp_snap["store_name"], gh_owner, gh_repo
517 )
519 if not validation["success"]:
520 return (validation["error"]["message"], 400)
522 if launchpad.is_snap_building(lp_snap["store_name"]):
523 launchpad.cancel_snap_builds(lp_snap["store_name"])
525 launchpad.build_snap(lp_snap["store_name"])
527 return ("", 204)
530@login_required
531def get_update_gh_webhooks(snap_name):
532 details = dashboard.get_snap_info(flask.session, snap_name)
534 lp_snap = launchpad.get_snap_by_store_name(details["snap_name"])
536 if not lp_snap:
537 flask.flash(
538 "This snap is not linked with a GitHub repository", "negative"
539 )
541 return flask.redirect(
542 flask.url_for(".get_settings", snap_name=snap_name)
543 )
545 github = GitHub(flask.session.get("github_auth_secret"))
547 try:
548 github.get_user()
549 except Unauthorized:
550 return flask.redirect(f"/github/auth?back={flask.request.path}")
552 gh_link = lp_snap["git_repository_url"][19:]
553 gh_owner, gh_repo = gh_link.split("/")
555 try:
556 # Remove old BSI webhook if present
557 old_url = (
558 f"https://build.snapcraft.io/{gh_owner}/{gh_repo}/webhook/notify"
559 )
560 old_hook = github.get_hook_by_url(gh_owner, gh_repo, old_url)
562 if old_hook:
563 github.remove_hook(
564 gh_owner,
565 gh_repo,
566 old_hook["id"],
567 )
569 # Remove current hook
570 github_hook_url = (
571 f"{GITHUB_WEBHOOK_HOST_URL}api/{snap_name}/webhook/notify"
572 )
573 snapcraft_hook = github.get_hook_by_url(
574 gh_owner, gh_repo, github_hook_url
575 )
577 if snapcraft_hook:
578 github.remove_hook(
579 gh_owner,
580 gh_repo,
581 snapcraft_hook["id"],
582 )
584 # Create webhook in the repo
585 github.create_hook(gh_owner, gh_repo, github_hook_url)
586 except HTTPError:
587 flask.flash(
588 "The GitHub Webhook could not be created. "
589 "Please try again or check your permissions over the repository.",
590 "caution",
591 )
592 else:
593 flask.flash("The webhook has been created successfully", "positive")
595 return flask.redirect(flask.url_for(".get_settings", snap_name=snap_name))