1# Copyright (C) 2020 The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""Module to check updates from crates.io.""" 15 16import json 17import os 18# pylint: disable=g-importing-member 19from pathlib import Path 20import re 21import shutil 22import tempfile 23import urllib.request 24 25import archive_utils 26from base_updater import Updater 27# pylint: disable=import-error 28import metadata_pb2 # type: ignore 29import updater_utils 30 31CRATES_IO_URL_PATTERN: str = (r"^https:\/\/crates.io\/crates\/([-\w]+)") 32 33CRATES_IO_URL_RE: re.Pattern = re.compile(CRATES_IO_URL_PATTERN) 34 35ALPHA_BETA_PATTERN: str = (r"^.*[0-9]+\.[0-9]+\.[0-9]+-(alpha|beta).*") 36 37ALPHA_BETA_RE: re.Pattern = re.compile(ALPHA_BETA_PATTERN) 38 39VERSION_PATTERN: str = (r"([0-9]+)\.([0-9]+)\.([0-9]+)") 40 41VERSION_MATCHER: re.Pattern = re.compile(VERSION_PATTERN) 42 43DESCRIPTION_PATTERN: str = (r"^description *= *(\".+\")") 44 45DESCRIPTION_MATCHER: re.Pattern = re.compile(DESCRIPTION_PATTERN) 46 47 48class CratesUpdater(Updater): 49 """Updater for crates.io packages.""" 50 51 download_url: str 52 package: str 53 package_dir: str 54 temp_file: tempfile.NamedTemporaryFile 55 56 def is_supported_url(self) -> bool: 57 if self._old_url.type != metadata_pb2.URL.HOMEPAGE: 58 return False 59 match = CRATES_IO_URL_RE.match(self._old_url.value) 60 if match is None: 61 return False 62 self.package = match.group(1) 63 return True 64 65 def _get_version_numbers(self, version: str) -> (int, int, int): 66 match = VERSION_MATCHER.match(version) 67 if match is not None: 68 return tuple(int(match.group(i)) for i in range(1, 4)) 69 return (0, 0, 0) 70 71 def _is_newer_version(self, prev_version: str, prev_id: int, 72 check_version: str, check_id: int): 73 """Return true if check_version+id is newer than prev_version+id.""" 74 return ((self._get_version_numbers(check_version), check_id) > 75 (self._get_version_numbers(prev_version), prev_id)) 76 77 def _find_latest_non_test_version(self) -> None: 78 url = "https://crates.io/api/v1/crates/{}/versions".format(self.package) 79 with urllib.request.urlopen(url) as request: 80 data = json.loads(request.read().decode()) 81 last_id = 0 82 self._new_ver = "" 83 for v in data["versions"]: 84 version = v["num"] 85 if (not v["yanked"] and not ALPHA_BETA_RE.match(version) and 86 self._is_newer_version( 87 self._new_ver, last_id, version, int(v["id"]))): 88 last_id = int(v["id"]) 89 self._new_ver = version 90 self.download_url = "https://crates.io" + v["dl_path"] 91 92 def check(self) -> None: 93 """Checks crates.io and returns whether a new version is available.""" 94 url = "https://crates.io/api/v1/crates/" + self.package 95 with urllib.request.urlopen(url) as request: 96 data = json.loads(request.read().decode()) 97 self._new_ver = data["crate"]["max_version"] 98 # Skip d.d.d-{alpha,beta}* versions 99 if ALPHA_BETA_RE.match(self._new_ver): 100 print("Ignore alpha or beta release: {}-{}." 101 .format(self.package, self._new_ver)) 102 self._find_latest_non_test_version() 103 else: 104 url = url + "/" + self._new_ver 105 with urllib.request.urlopen(url) as request: 106 data = json.loads(request.read().decode()) 107 self.download_url = "https://crates.io" + data["version"]["dl_path"] 108 109 def use_current_as_latest(self): 110 Updater.use_current_as_latest(self) 111 # A shortcut to use the static download path. 112 self.download_url = "https://static.crates.io/crates/{}/{}-{}.crate".format( 113 self.package, self.package, self._new_ver) 114 115 def update(self) -> None: 116 """Updates the package. 117 118 Has to call check() before this function. 119 """ 120 try: 121 temporary_dir = archive_utils.download_and_extract(self.download_url) 122 self.package_dir = archive_utils.find_archive_root(temporary_dir) 123 self.temp_file = tempfile.NamedTemporaryFile() 124 updater_utils.replace_package(self.package_dir, self._proj_path, 125 self.temp_file.name) 126 self.check_for_errors() 127 finally: 128 urllib.request.urlcleanup() 129 130 def rollback(self) -> bool: 131 # Only rollback if we have already swapped, 132 # which we denote by writing to this file. 133 if os.fstat(self.temp_file.fileno()).st_size > 0: 134 tmp_dir = tempfile.TemporaryDirectory() 135 shutil.move(self._proj_path, tmp_dir.name) 136 shutil.move(self.package_dir, self._proj_path) 137 shutil.move(Path(tmp_dir.name) / self.package, self.package_dir) 138 return True 139 return False 140 141 # pylint: disable=no-self-use 142 def update_metadata(self, metadata: metadata_pb2.MetaData, 143 full_path: Path) -> None: 144 """Updates METADATA content.""" 145 # copy only HOMEPAGE url, and then add new ARCHIVE url. 146 new_url_list = [] 147 for url in metadata.third_party.url: 148 if url.type == metadata_pb2.URL.HOMEPAGE: 149 new_url_list.append(url) 150 new_url = metadata_pb2.URL() 151 new_url.type = metadata_pb2.URL.ARCHIVE 152 new_url.value = "https://static.crates.io/crates/{}/{}-{}.crate".format( 153 metadata.name, metadata.name, metadata.third_party.version) 154 new_url_list.append(new_url) 155 del metadata.third_party.url[:] 156 metadata.third_party.url.extend(new_url_list) 157 # copy description from Cargo.toml to METADATA 158 cargo_toml = os.path.join(full_path, "Cargo.toml") 159 description = self._get_cargo_description(cargo_toml) 160 if description and description != metadata.description: 161 print("New METADATA description:", description) 162 metadata.description = description 163 164 def check_for_errors(self) -> None: 165 # Check for .rej patches from failing to apply patches. 166 # If this has too many false positives, we could either 167 # check if the files are modified by patches or somehow 168 # track which files existed before the patching. 169 rejects = list(self._proj_path.glob('**/*.rej')) 170 if len(rejects) > 0: 171 print("Error: Found patch reject files: %s" % str(rejects)) 172 self._has_errors = True 173 # Check for Cargo errors embedded in Android.bp. 174 # Note that this should stay in sync with cargo2android.py. 175 with open('%s/Android.bp' % self._proj_path, 'r') as bp_file: 176 for line in bp_file: 177 if line.strip() == "Errors in cargo.out:": 178 print("Error: Found Cargo errors in Android.bp") 179 self._has_errors = True 180 return 181 182 def _toml2str(self, line: str) -> str: 183 """Convert a quoted toml string to a Python str without quotes.""" 184 if line.startswith("\"\"\""): 185 return "" # cannot handle broken multi-line description 186 # TOML string escapes: \b \t \n \f \r \" \\ (no unicode escape) 187 line = line[1:-1].replace("\\\\", "\n").replace("\\b", "") 188 line = line.replace("\\t", " ").replace("\\n", " ").replace("\\f", " ") 189 line = line.replace("\\r", "").replace("\\\"", "\"").replace("\n", "\\") 190 # replace a unicode quotation mark, used in the libloading crate 191 return line.replace("’", "'").strip() 192 193 def _get_cargo_description(self, cargo_toml: str) -> str: 194 """Return the description in Cargo.toml or empty string.""" 195 if os.path.isfile(cargo_toml) and os.access(cargo_toml, os.R_OK): 196 with open(cargo_toml, "r") as toml_file: 197 for line in toml_file: 198 match = DESCRIPTION_MATCHER.match(line) 199 if match: 200 return self._toml2str(match.group(1)) 201 return "" 202