tools/tryselect/lando.py

   1 # This Source Code Form is subject to the terms of the Mozilla Public
   2 # License, v. 2.0. If a copy of the MPL was not distributed with this
   3 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   4
   5 """Implements Auth0 Device Code flow and Lando try submission.
   6
   7 See https://auth0.com/blog/securing-a-python-cli-application-with-auth0/ for more.
   8 """
   9
  10 from __future__ import annotations
  11
  12 import base64
  13 import configparser
  14 import json
  15 import os
  16 import time
  17 import webbrowser
  18 from dataclasses import (
  19     dataclass,
  20     field,
  21 )
  22 from pathlib import Path
  23 from typing import (
  24     List,
  25     Optional,
  26     Tuple,
  27     Union,
  28 )
  29
  30 import requests
  31 from mach.util import get_state_dir
  32 from mozbuild.base import MozbuildObject
  33 from mozversioncontrol import (
  34     GitRepository,
  35     HgRepository,
  36 )
  37
  38 from .task_config import (
  39     try_config_commit,
  40 )
  41
  42 TOKEN_FILE = (
  43     Path(get_state_dir(specific_to_topsrcdir=False)) / "lando_auth0_user_token.json"
  44 )
  45
  46 # The supported variants of `Repository` for this workflow.
  47 SupportedVcsRepository = Union[GitRepository, HgRepository]
  48
  49 here = os.path.abspath(os.path.dirname(__file__))
  50 build = MozbuildObject.from_environment(cwd=here)
  51
  52
  53 def convert_bytes_patch_to_base64(patch_bytes: bytes) -> str:
  54     """Return a base64 encoded `str` representing the passed `bytes` patch."""
  55     return base64.b64encode(patch_bytes).decode("ascii")
  56
  57
  58 def load_token_from_disk() -> Optional[dict]:
  59     """Load and validate an existing Auth0 token from disk.
  60
  61     Return the token as a `dict` if it can be validated, or return `None`
  62     if any error was encountered.
  63     """
  64     if not TOKEN_FILE.exists():
  65         print("No existing Auth0 token found.")
  66         return None
  67
  68     try:
  69         user_token = json.loads(TOKEN_FILE.read_bytes())
  70     except json.JSONDecodeError:
  71         print("Existing Auth0 token could not be decoded as JSON.")
  72         return None
  73
  74     return user_token
  75
  76
  77 def get_stack_info(vcs: SupportedVcsRepository) -> Tuple[str, List[str]]:
  78     """Retrieve information about the current stack for submission via Lando.
  79
  80     Returns a tuple of the current public base commit as a Mercurial SHA,
  81     and a list of ordered base64 encoded patches.
  82     """
  83     base_commit = vcs.base_ref_as_hg()
  84     if not base_commit:
  85         raise ValueError(
  86             "Could not determine base Mercurial commit hash for submission."
  87         )
  88     print("Using", base_commit, "as the hg base commit.")
  89
  90     # Reuse the base revision when on Mercurial to avoid multiple calls to `hg log`.
  91     branch_nodes_kwargs = {}
  92     if isinstance(vcs, HgRepository):
  93         branch_nodes_kwargs["base_ref"] = base_commit
  94
  95     nodes = vcs.get_branch_nodes(**branch_nodes_kwargs)
  96     if not nodes:
  97         raise ValueError("Could not find any commit hashes for submission.")
  98     elif len(nodes) == 1:
  99         print("Submitting a single try config commit.")
 100     elif len(nodes) == 2:
 101         print("Submitting 1 node and the try commit.")
 102     else:
 103         print("Submitting stack of", len(nodes) - 1, "nodes and the try commit.")
 104
 105     patches = vcs.get_commit_patches(nodes)
 106     base64_patches = [
 107         convert_bytes_patch_to_base64(patch_bytes) for patch_bytes in patches
 108     ]
 109     print("Patches gathered for submission.")
 110
 111     return base_commit, base64_patches
 112
 113
 114 @dataclass
 115 class Auth0Config:
 116     """Helper class to interact with Auth0."""
 117
 118     domain: str
 119     client_id: str
 120     audience: str
 121     scope: str
 122     algorithms: list[str] = field(default_factory=lambda: ["RS256"])
 123
 124     @property
 125     def base_url(self) -> str:
 126         """Auth0 base URL."""
 127         return f"https://{self.domain}"
 128
 129     @property
 130     def device_code_url(self) -> str:
 131         """URL of the Device Code API endpoint."""
 132         return f"{self.base_url}/oauth/device/code"
 133
 134     @property
 135     def issuer(self) -> str:
 136         """Token issuer URL."""
 137         return f"{self.base_url}/"
 138
 139     @property
 140     def jwks_url(self) -> str:
 141         """URL of the JWKS file."""
 142         return f"{self.base_url}/.well-known/jwks.json"
 143
 144     @property
 145     def oauth_token_url(self) -> str:
 146         """URL of the OAuth Token endpoint."""
 147         return f"{self.base_url}/oauth/token"
 148
 149     def request_device_code(self) -> dict:
 150         """Request authorization from Auth0 using the Device Code Flow.
 151
 152         See https://auth0.com/docs/api/authentication#get-device-code for more.
 153         """
 154         response = requests.post(
 155             self.device_code_url,
 156             headers={"Content-Type": "application/x-www-form-urlencoded"},
 157             data={
 158                 "audience": self.audience,
 159                 "client_id": self.client_id,
 160                 "scope": self.scope,
 161             },
 162         )
 163
 164         response.raise_for_status()
 165
 166         return response.json()
 167
 168     def validate_token(self, user_token: dict) -> Optional[dict]:
 169         """Verify the given user token is valid.
 170
 171         Validate the ID token, and validate the access token's expiration claim.
 172         """
 173         # Import `auth0-python` here to avoid `ImportError` in tests, since
 174         # the `python-test` site won't have `auth0-python` installed.
 175         import jwt
 176         from auth0.authentication.token_verifier import (
 177             AsymmetricSignatureVerifier,
 178             TokenVerifier,
 179         )
 180         from auth0.exceptions import (
 181             TokenValidationError,
 182         )
 183
 184         signature_verifier = AsymmetricSignatureVerifier(self.jwks_url)
 185         token_verifier = TokenVerifier(
 186             audience=self.client_id,
 187             issuer=self.issuer,
 188             signature_verifier=signature_verifier,
 189         )
 190
 191         try:
 192             token_verifier.verify(user_token["id_token"])
 193         except TokenValidationError as e:
 194             print("Could not validate existing Auth0 ID token:", str(e))
 195             return None
 196
 197         decoded_access_token = jwt.decode(
 198             user_token["access_token"],
 199             algorithms=self.algorithms,
 200             options={"verify_signature": False},
 201         )
 202
 203         access_token_expiration = decoded_access_token["exp"]
 204
 205         # Assert that the access token isn't expired or expiring within a minute.
 206         if time.time() > access_token_expiration + 60:
 207             print("Access token is expired.")
 208             return None
 209
 210         user_token.update(
 211             jwt.decode(
 212                 user_token["id_token"],
 213                 algorithms=self.algorithms,
 214                 options={"verify_signature": False},
 215             )
 216         )
 217         print("Auth0 token validated.")
 218         return user_token
 219
 220     def device_authorization_flow(self) -> dict:
 221         """Perform the Device Authorization Flow.
 222
 223         See https://auth0.com/docs/get-started/authentication-and-authorization-flow/device-authorization-flow
 224         for more.
 225         """
 226         start = time.perf_counter()
 227
 228         device_code_data = self.request_device_code()
 229         print(
 230             "1. On your computer or mobile device navigate to:",
 231             device_code_data["verification_uri_complete"],
 232         )
 233         print("2. Enter the following code:", device_code_data["user_code"])
 234
 235         auth_msg = f"Auth0 token validation required at: {device_code_data['verification_uri_complete']}"
 236         build.notify(auth_msg)
 237
 238         try:
 239             webbrowser.open(device_code_data['verification_uri_complete'])
 240         except webbrowser.Error:
 241             print("Could not automatically open the web browser.")
 242
 243         device_code_lifetime_s = device_code_data["expires_in"]
 244
 245         # Print successive periods on the same line to avoid moving the link
 246         # while the user is trying to click it.
 247         print("Waiting...", end="", flush=True)
 248         while time.perf_counter() - start < device_code_lifetime_s:
 249             response = requests.post(
 250                 self.oauth_token_url,
 251                 data={
 252                     "client_id": self.client_id,
 253                     "device_code": device_code_data["device_code"],
 254                     "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
 255                     "scope": self.scope,
 256                 },
 257             )
 258             response_data = response.json()
 259
 260             if response.status_code == 200:
 261                 print("\nLogin successful.")
 262                 return response_data
 263
 264             if response_data["error"] not in ("authorization_pending", "slow_down"):
 265                 raise RuntimeError(response_data["error_description"])
 266
 267             time.sleep(device_code_data["interval"])
 268             print(".", end="", flush=True)
 269
 270         raise ValueError("Timed out waiting for Auth0 device code authentication!")
 271
 272     def get_token(self) -> dict:
 273         """Retrieve an access token for authentication.
 274
 275         If a cached token is found and can be confirmed to be valid, return it.
 276         Otherwise, perform the Device Code Flow authorization to request a new
 277         token, validate it and save it to disk.
 278         """
 279         # Load a cached token and validate it if one is available.
 280         cached_token = load_token_from_disk()
 281         user_token = self.validate_token(cached_token) if cached_token else None
 282
 283         # Login with the Device Authorization Flow if an existing token isn't found.
 284         if not user_token:
 285             new_token = self.device_authorization_flow()
 286             user_token = self.validate_token(new_token)
 287
 288         if not user_token:
 289             raise ValueError("Could not get an Auth0 token.")
 290
 291         # Save token to disk.
 292         with TOKEN_FILE.open("w") as f:
 293             json.dump(user_token, f, indent=2, sort_keys=True)
 294
 295         return user_token
 296
 297
 298 class LandoAPIException(Exception):
 299     """Raised when Lando throws an exception."""
 300
 301     def __init__(self, detail: Optional[str] = None):
 302         super().__init__(detail or "")
 303
 304
 305 @dataclass
 306 class LandoAPI:
 307     """Helper class to interact with Lando-API."""
 308
 309     access_token: str
 310     api_url: str
 311
 312     @property
 313     def lando_try_api_url(self) -> str:
 314         """URL of the Lando Try endpoint."""
 315         return f"https://{self.api_url}/try/patches"
 316
 317     @property
 318     def api_headers(self) -> dict[str, str]:
 319         """Headers for use accessing and authenticating against the API."""
 320         return {
 321             "Authorization": f"Bearer {self.access_token}",
 322             "Content-Type": "application/json",
 323         }
 324
 325     @classmethod
 326     def from_lando_config_file(cls, config_path: Path, section: str) -> LandoAPI:
 327         """Build a `LandoConfig` from `section` in the file at `config_path`."""
 328         if not config_path.exists():
 329             raise ValueError(f"Could not find a Lando config file at `{config_path}`.")
 330
 331         lando_ini_contents = config_path.read_text()
 332
 333         parser = configparser.ConfigParser(delimiters="=")
 334         parser.read_string(lando_ini_contents)
 335
 336         if not parser.has_section(section):
 337             raise ValueError(f"Lando config file does not have a {section} section.")
 338
 339         auth0 = Auth0Config(
 340             domain=parser.get(section, "auth0_domain"),
 341             client_id=parser.get(section, "auth0_client_id"),
 342             audience=parser.get(section, "auth0_audience"),
 343             scope=parser.get(section, "auth0_scope"),
 344         )
 345
 346         token = auth0.get_token()
 347
 348         return LandoAPI(
 349             api_url=parser.get(section, "api_domain"),
 350             access_token=token["access_token"],
 351         )
 352
 353     def post(self, url: str, body: dict) -> dict:
 354         """Make a POST request to Lando."""
 355         response = requests.post(url, headers=self.api_headers, json=body)
 356
 357         try:
 358             response_json = response.json()
 359         except json.JSONDecodeError:
 360             # If the server didn't send back a valid JSON object, raise a stack
 361             # trace to the terminal which includes error details.
 362             response.raise_for_status()
 363
 364             # Raise `ValueError` if the response wasn't JSON and we didn't raise
 365             # from an invalid status.
 366             raise LandoAPIException(
 367                 detail="Response was not valid JSON yet status was valid."
 368             )
 369
 370         if response.status_code >= 400:
 371             raise LandoAPIException(detail=response_json["detail"])
 372
 373         return response_json
 374
 375     def post_try_push_patches(
 376         self,
 377         patches: List[str],
 378         patch_format: str,
 379         base_commit: str,
 380     ) -> dict:
 381         """Send try push contents to Lando.
 382
 383         Send the list of base64-encoded `patches` in `patch_format` to Lando, to be applied to
 384         the Mercurial `base_commit`, using the Auth0 `access_token` for authorization.
 385         """
 386         request_json_body = {
 387             "base_commit": base_commit,
 388             "patch_format": patch_format,
 389             "patches": patches,
 390         }
 391
 392         print("Submitting patches to Lando.")
 393         response_json = self.post(self.lando_try_api_url, request_json_body)
 394
 395         return response_json
 396
 397
 398 def push_to_lando_try(vcs: SupportedVcsRepository, commit_message: str):
 399     """Push a set of patches to Lando's try endpoint."""
 400     # Map `Repository` subclasses to the `patch_format` value Lando expects.
 401     PATCH_FORMAT_STRING_MAPPING = {
 402         GitRepository: "git-format-patch",
 403         HgRepository: "hgexport",
 404     }
 405     patch_format = PATCH_FORMAT_STRING_MAPPING.get(type(vcs))
 406     if not patch_format:
 407         # Other VCS types (namely `src`) are unsupported.
 408         raise ValueError(f"Try push via Lando is not supported for `{vcs.name}`.")
 409
 410     # Use Lando Prod unless the `LANDO_TRY_USE_DEV` environment variable is defined.
 411     lando_config_section = (
 412         "lando-prod" if not os.getenv("LANDO_TRY_USE_DEV") else "lando-dev"
 413     )
 414
 415     # Load Auth0 config from `.lando.ini`.
 416     lando_ini_path = Path(vcs.path) / ".lando.ini"
 417     lando_api = LandoAPI.from_lando_config_file(lando_ini_path, lando_config_section)
 418
 419     # Get the time when the push was initiated, not including Auth0 login time.
 420     push_start_time = time.perf_counter()
 421
 422     with try_config_commit(vcs, commit_message):
 423         try:
 424             base_commit, patches = get_stack_info(vcs)
 425         except ValueError as exc:
 426             error_msg = "abort: error gathering patches for submission."
 427             print(error_msg)
 428             print(str(exc))
 429             build.notify(error_msg)
 430             return
 431
 432         try:
 433             # Make the try request to Lando.
 434             response_json = lando_api.post_try_push_patches(
 435                 patches, patch_format, base_commit
 436             )
 437         except LandoAPIException as exc:
 438             error_msg = "abort: error submitting patches to Lando."
 439             print(error_msg)
 440             print(str(exc))
 441             build.notify(error_msg)
 442             return
 443
 444     duration = round(time.perf_counter() - push_start_time, ndigits=2)
 445
 446     job_id = response_json["id"]
 447     success_msg = (
 448         f"Lando try submission success, took {duration} seconds. "
 449         f"Landing job id: {job_id}."
 450     )
 451     print(success_msg)
 452     build.notify(success_msg)