| # Copyright (C) 2020 Savoir-Faire Linux |
| # |
| # SPDX-License-Identifier: GPL-2.0-only |
| # |
| """ |
| BitBake 'Fetch' npm implementation |
| |
| npm fetcher support the SRC_URI with format of: |
| SRC_URI = "npm://some.registry.url;OptionA=xxx;OptionB=xxx;..." |
| |
| Supported SRC_URI options are: |
| |
| - package |
| The npm package name. This is a mandatory parameter. |
| |
| - version |
| The npm package version. This is a mandatory parameter. |
| |
| - sha512sum / sha256sum / sha384sum / sha1sum |
| The expected checksum of the downloaded tarball. Exactly one is |
| mandatory: the registry cannot be trusted to supply its own tamper |
| detection, so the checksum must come from the recipe. |
| |
| - downloadfilename |
| Specifies the filename used when storing the downloaded file. |
| |
| - destsuffix |
| Specifies the directory to use to unpack the package (default: npm). |
| """ |
| |
| import base64 |
| import json |
| import os |
| import re |
| import shlex |
| import tempfile |
| import bb |
| from bb.fetch import Fetch |
| from bb.fetch import FetchError |
| from bb.fetch import FetchMethod |
| from bb.fetch import MalformedUrl |
| from bb.fetch import MissingParameterError |
| from bb.fetch import ParameterError |
| from bb.fetch import URI |
| from bb.fetch import check_network_access |
| from bb.fetch import runfetchcmd |
| from bb.utils import is_semver |
| |
| # Preference order matches strength; the first one present in SRC_URI wins. |
| CHECKSUM_PARMS = ("sha512sum", "sha256sum", "sha384sum", "sha1sum") |
| |
| def npm_package(package): |
| """Convert the npm package name to remove unsupported character""" |
| # For scoped package names ('@user/package') the '/' is replaced by a '-'. |
| # This is similar to what 'npm pack' does, but 'npm pack' also strips the |
| # leading '@', which can lead to ambiguous package names. |
| name = re.sub("/", "-", package) |
| name = name.lower() |
| name = re.sub(r"[^\-a-z0-9@]", "", name) |
| name = name.strip("-") |
| return name |
| |
| |
| def npm_filename(package, version): |
| """Get the filename of a npm package""" |
| return npm_package(package) + "-" + version + ".tgz" |
| |
| def npm_localfile(package, version=None): |
| """Get the local filename of a npm package""" |
| if version is not None: |
| filename = npm_filename(package, version) |
| else: |
| filename = package |
| return os.path.join("npm2", filename) |
| |
| def npm_integrity(integrity): |
| """ |
| Get the checksum name and expected value from the subresource integrity |
| https://www.w3.org/TR/SRI/ |
| """ |
| algo, value = integrity.split("-", maxsplit=1) |
| return "%ssum" % algo, base64.b64decode(value).hex() |
| |
| def npm_unpack(tarball, destdir, d): |
| """Unpack a npm tarball""" |
| bb.utils.mkdirhier(destdir) |
| cmd = ['tar', '--extract', '--gzip', '--file=%s' % tarball, '--no-same-owner', '--delay-directory-restore', '--strip-components=1'] |
| runfetchcmd(cmd, d, workdir=destdir) |
| runfetchcmd(['chmod', '-R', '+X', destdir], d, quiet=True, workdir=destdir) |
| |
| class NpmEnvironment(object): |
| """ |
| Using a npm config file seems more reliable than using cli arguments. |
| This class allows to create a controlled environment for npm commands. |
| """ |
| def __init__(self, d, configs=[], npmrc=None): |
| self.d = d |
| |
| self.user_config = tempfile.NamedTemporaryFile(mode="w", buffering=1) |
| |
| hn = self._home_npmrc(d) |
| if hn is not None: |
| with open(hn, 'r') as hnf: |
| self.user_config.write(hnf.read()) |
| |
| for key, value in configs: |
| self.user_config.write("%s=%s\n" % (key, value)) |
| |
| if npmrc: |
| self.global_config_name = npmrc |
| else: |
| self.global_config_name = "/dev/null" |
| |
| def __del__(self): |
| if self.user_config: |
| self.user_config.close() |
| |
| def _home_npmrc(self, d): |
| """Function to return user's HOME .npmrc file (or None if it doesn't exist)""" |
| home_npmrc_file = os.path.join(os.environ.get("HOME"), ".npmrc") |
| if d.getVar("BB_USE_HOME_NPMRC") == "1" and os.path.exists(home_npmrc_file): |
| bb.warn(f"BB_USE_HOME_NPMRC flag set and valid .npmrc detected - "\ |
| f"npm fetcher will use {home_npmrc_file}") |
| return home_npmrc_file |
| return None |
| |
| def run(self, cmd, args=None, configs=None, workdir=None): |
| """Run npm command in a controlled environment""" |
| with tempfile.TemporaryDirectory() as tmpdir: |
| d = bb.data.createCopy(self.d) |
| d.setVar("PATH", d.getVar("PATH")) # PATH might contain $HOME - evaluate it before patching |
| d.setVar("HOME", tmpdir) |
| |
| if not workdir: |
| workdir = tmpdir |
| |
| def _run(cmd): |
| extraenv = {} |
| extraenv['NPM_CONFIG_USERCONFIG'] = self.user_config.name |
| extraenv['NPM_CONFIG_GLOBALCONFIG'] = self.global_config_name |
| return runfetchcmd(cmd, d, workdir=workdir, extraenv=extraenv) |
| |
| if configs: |
| bb.warn("Use of configs argument of NpmEnvironment.run() function" |
| " is deprecated. Please use args argument instead.") |
| for key, value in configs: |
| cmd.append('--%s=%s' % (key, value)) |
| |
| if args: |
| for key, value in args: |
| cmd.append('--%s=%s' % (key, value)) |
| |
| return _run(cmd) |
| |
| class Npm(FetchMethod): |
| """Class to fetch a package from a npm registry""" |
| |
| def supports(self, ud, d): |
| """Check if a given url can be fetched with npm""" |
| return ud.type in ["npm"] |
| |
| def urldata_init(self, ud, d): |
| """Init npm specific variables within url data""" |
| ud.package = None |
| ud.version = None |
| ud.registry = None |
| |
| # Get the 'package' parameter |
| if "package" in ud.parm: |
| ud.package = ud.parm.get("package") |
| |
| if not ud.package: |
| raise MissingParameterError("Parameter 'package' required", ud.url) |
| |
| # Get the 'version' parameter |
| if "version" in ud.parm: |
| ud.version = ud.parm.get("version") |
| |
| if not ud.version: |
| raise MissingParameterError("Parameter 'version' required", ud.url) |
| |
| if ud.version == "latest": |
| raise ParameterError( |
| "Version 'latest' is not reproducible; specify an exact semver version", |
| ud.url) |
| |
| if not is_semver(ud.version): |
| raise ParameterError("Invalid 'version' parameter", ud.url) |
| |
| # Extract the 'registry' part of the url |
| ud.registry = re.sub(r"^npm://", "https://", ud.url.split(";")[0]) |
| if not ud.url.split(";")[0][len("npm://"):]: |
| raise MalformedUrl(ud.url) |
| |
| # A checksum is mandatory: the registry cannot be trusted to supply |
| # its own tamper detection, so the recipe must commit to one. |
| for cp in CHECKSUM_PARMS: |
| if cp in ud.parm: |
| ud.checksum_name = cp |
| ud.checksum_expected = ud.parm[cp] |
| break |
| else: |
| raise MissingParameterError( |
| "Missing checksum for npm package '%s@%s': a compromised " |
| "registry could otherwise serve a tampered tarball undetected. " |
| "Add one of %s to SRC_URI." % (ud.package, ud.version, ", ".join(CHECKSUM_PARMS)), |
| ud.url) |
| |
| # Using the 'downloadfilename' parameter as local filename |
| # or the npm package name. |
| if "downloadfilename" in ud.parm: |
| ud.localfile = npm_localfile(ud.parm["downloadfilename"]) |
| else: |
| ud.localfile = npm_localfile(ud.package, ud.version) |
| |
| # Get the base 'npm' command |
| ud.basecmd = shlex.split(d.getVar("FETCHCMD_npm") or "") or ["npm"] |
| |
| # This fetcher resolves a URI from a npm package name and version and |
| # then forwards it to a proxy fetcher. A resolve file containing the |
| # resolved URI is created to avoid unwanted network access (if the file |
| # already exists). The management of the donestamp file, the lockfile |
| # and the checksums are forwarded to the proxy fetcher. |
| ud.proxy = None |
| ud.needdonestamp = False |
| ud.resolvefile = self.localpath(ud, d) + ".resolved" |
| |
| def _resolve_proxy_url(self, ud, d): |
| """Resolve the tarball URL from the registry and cache it without any checksum. |
| |
| Checksums must never be sourced from the registry: a compromised registry |
| controls both the tarball and its advertised hash, so any checksum obtained |
| there provides no tamper detection. Checksums are applied in _setup_proxy |
| from the recipe-provided SRC_URI parameters instead. |
| """ |
| def _npm_view(): |
| args = [] |
| args.append(("json", "true")) |
| args.append(("registry", ud.registry)) |
| pkgver = ud.package + "@" + ud.version |
| cmd = ud.basecmd + ['view', pkgver] |
| env = NpmEnvironment(d) |
| check_network_access(d, cmd, ud.registry) |
| view_string = env.run(cmd, args=args) |
| |
| if not view_string: |
| raise FetchError("Unavailable package %s" % pkgver, ud.url) |
| |
| try: |
| view = json.loads(view_string) |
| except json.JSONDecodeError as e: |
| raise FetchError("Invalid view from npm: %s" % str(e), ud.url) |
| |
| error = view.get("error") |
| if error is not None: |
| raise FetchError(error.get("summary") or str(error), ud.url) |
| |
| if ud.version != view.get("version"): |
| raise ParameterError("Invalid 'version' parameter", ud.url) |
| |
| return view |
| |
| view = _npm_view() |
| tarball_url = view.get("dist", {}).get("tarball") |
| |
| if tarball_url is None: |
| raise FetchError("Invalid 'dist.tarball' in view", ud.url) |
| |
| bb.utils.mkdirhier(os.path.dirname(ud.resolvefile)) |
| with open(ud.resolvefile, "w") as f: |
| f.write(tarball_url) |
| |
| def _setup_proxy(self, ud, d): |
| if ud.proxy is None: |
| if not os.path.exists(ud.resolvefile): |
| self._resolve_proxy_url(ud, d) |
| |
| with open(ud.resolvefile, "r") as f: |
| tarball_url = f.read().strip() |
| |
| uri = URI(tarball_url) |
| # Discard any params that may have been embedded in the stored URL |
| # (e.g. from an npmsw-written .resolved file) and rebuild from |
| # scratch so that registry-sourced checksums can never flow through. |
| uri.params = {} |
| uri.params["downloadfilename"] = ud.localfile |
| |
| # Inject the recipe-provided checksum into the proxy URL. |
| # Checksums from the remote registry are never used; only the |
| # value validated in urldata_init, sourced from the recipe, is |
| # trusted. |
| uri.params[ud.checksum_name] = ud.checksum_expected |
| |
| # Avoid conflicts between the environment data and: |
| # - the proxy url checksum |
| data = bb.data.createCopy(d) |
| data.delVarFlags("SRC_URI") |
| ud.proxy = Fetch([str(uri)], data) |
| |
| def _get_proxy_method(self, ud, d): |
| self._setup_proxy(ud, d) |
| proxy_url = ud.proxy.urls[0] |
| proxy_ud = ud.proxy.ud[proxy_url] |
| proxy_d = ud.proxy.d |
| proxy_ud.setup_localpath(proxy_d) |
| return proxy_ud.method, proxy_ud, proxy_d |
| |
| def verify_donestamp(self, ud, d): |
| """Verify the donestamp file""" |
| proxy_m, proxy_ud, proxy_d = self._get_proxy_method(ud, d) |
| return proxy_m.verify_donestamp(proxy_ud, proxy_d) |
| |
| def update_donestamp(self, ud, d): |
| """Update the donestamp file""" |
| proxy_m, proxy_ud, proxy_d = self._get_proxy_method(ud, d) |
| proxy_m.update_donestamp(proxy_ud, proxy_d) |
| |
| def need_update(self, ud, d): |
| """Force a fetch, even if localpath exists ?""" |
| if not os.path.exists(ud.resolvefile): |
| return True |
| proxy_m, proxy_ud, proxy_d = self._get_proxy_method(ud, d) |
| return proxy_m.need_update(proxy_ud, proxy_d) |
| |
| def try_mirrors(self, fetch, ud, d, mirrors): |
| """Try to use a mirror""" |
| proxy_m, proxy_ud, proxy_d = self._get_proxy_method(ud, d) |
| return proxy_m.try_mirrors(fetch, proxy_ud, proxy_d, mirrors) |
| |
| def download(self, ud, d): |
| """Fetch url""" |
| self._setup_proxy(ud, d) |
| ud.proxy.download() |
| |
| def unpack(self, ud, rootdir, d): |
| """Unpack the downloaded archive""" |
| destsuffix = ud.parm.get("destsuffix", "npm") |
| destdir = os.path.join(rootdir, destsuffix) |
| npm_unpack(ud.localpath, destdir, d) |
| ud.unpack_tracer.unpack("npm", destdir) |
| |
| def clean(self, ud, d): |
| """Clean any existing full or partial download""" |
| if os.path.exists(ud.resolvefile): |
| self._setup_proxy(ud, d) |
| ud.proxy.clean() |
| bb.utils.remove(ud.resolvefile) |
| |
| def done(self, ud, d): |
| """Is the download done ?""" |
| if not os.path.exists(ud.resolvefile): |
| return False |
| proxy_m, proxy_ud, proxy_d = self._get_proxy_method(ud, d) |
| return proxy_m.done(proxy_ud, proxy_d) |
