# 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.fetch2 import Fetch
from bb.fetch2 import FetchError
from bb.fetch2 import FetchMethod
from bb.fetch2 import MalformedUrl
from bb.fetch2 import MissingParameterError
from bb.fetch2 import ParameterError
from bb.fetch2 import URI
from bb.fetch2 import check_network_access
from bb.fetch2 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)