| """ |
| Wheel command-line utility. |
| """ |
| |
| import os |
| import hashlib |
| import sys |
| import json |
| |
| from glob import iglob |
| from .. import signatures |
| from ..util import (urlsafe_b64decode, urlsafe_b64encode, native, binary, |
| matches_requirement) |
| from ..install import WheelFile, VerifyingZipFile |
| from ..paths import get_install_command |
| |
| def require_pkgresources(name): |
| try: |
| import pkg_resources |
| except ImportError: |
| raise RuntimeError("'{0}' needs pkg_resources (part of setuptools).".format(name)) |
| |
| import argparse |
| |
| class WheelError(Exception): pass |
| |
| # For testability |
| def get_keyring(): |
| try: |
| from ..signatures import keys |
| import keyring |
| assert keyring.get_keyring().priority |
| except (ImportError, AssertionError): |
| raise WheelError("Install wheel[signatures] (requires keyring, keyrings.alt, pyxdg) for signatures.") |
| return keys.WheelKeys, keyring |
| |
| def keygen(get_keyring=get_keyring): |
| """Generate a public/private key pair.""" |
| WheelKeys, keyring = get_keyring() |
| |
| ed25519ll = signatures.get_ed25519ll() |
| |
| wk = WheelKeys().load() |
| |
| keypair = ed25519ll.crypto_sign_keypair() |
| vk = native(urlsafe_b64encode(keypair.vk)) |
| sk = native(urlsafe_b64encode(keypair.sk)) |
| kr = keyring.get_keyring() |
| kr.set_password("wheel", vk, sk) |
| sys.stdout.write("Created Ed25519 keypair with vk={0}\n".format(vk)) |
| sys.stdout.write("in {0!r}\n".format(kr)) |
| |
| sk2 = kr.get_password('wheel', vk) |
| if sk2 != sk: |
| raise WheelError("Keyring is broken. Could not retrieve secret key.") |
| |
| sys.stdout.write("Trusting {0} to sign and verify all packages.\n".format(vk)) |
| wk.add_signer('+', vk) |
| wk.trust('+', vk) |
| wk.save() |
| |
| def sign(wheelfile, replace=False, get_keyring=get_keyring): |
| """Sign a wheel""" |
| WheelKeys, keyring = get_keyring() |
| |
| ed25519ll = signatures.get_ed25519ll() |
| |
| wf = WheelFile(wheelfile, append=True) |
| wk = WheelKeys().load() |
| |
| name = wf.parsed_filename.group('name') |
| sign_with = wk.signers(name)[0] |
| sys.stdout.write("Signing {0} with {1}\n".format(name, sign_with[1])) |
| |
| vk = sign_with[1] |
| kr = keyring.get_keyring() |
| sk = kr.get_password('wheel', vk) |
| keypair = ed25519ll.Keypair(urlsafe_b64decode(binary(vk)), |
| urlsafe_b64decode(binary(sk))) |
| |
| |
| record_name = wf.distinfo_name + '/RECORD' |
| sig_name = wf.distinfo_name + '/RECORD.jws' |
| if sig_name in wf.zipfile.namelist(): |
| raise WheelError("Wheel is already signed.") |
| record_data = wf.zipfile.read(record_name) |
| payload = {"hash":"sha256=" + native(urlsafe_b64encode(hashlib.sha256(record_data).digest()))} |
| sig = signatures.sign(payload, keypair) |
| wf.zipfile.writestr(sig_name, json.dumps(sig, sort_keys=True)) |
| wf.zipfile.close() |
| |
| def unsign(wheelfile): |
| """ |
| Remove RECORD.jws from a wheel by truncating the zip file. |
| |
| RECORD.jws must be at the end of the archive. The zip file must be an |
| ordinary archive, with the compressed files and the directory in the same |
| order, and without any non-zip content after the truncation point. |
| """ |
| vzf = VerifyingZipFile(wheelfile, "a") |
| info = vzf.infolist() |
| if not (len(info) and info[-1].filename.endswith('/RECORD.jws')): |
| raise WheelError("RECORD.jws not found at end of archive.") |
| vzf.pop() |
| vzf.close() |
| |
| def verify(wheelfile): |
| """Verify a wheel. |
| |
| The signature will be verified for internal consistency ONLY and printed. |
| Wheel's own unpack/install commands verify the manifest against the |
| signature and file contents. |
| """ |
| wf = WheelFile(wheelfile) |
| sig_name = wf.distinfo_name + '/RECORD.jws' |
| sig = json.loads(native(wf.zipfile.open(sig_name).read())) |
| verified = signatures.verify(sig) |
| sys.stderr.write("Signatures are internally consistent.\n") |
| sys.stdout.write(json.dumps(verified, indent=2)) |
| sys.stdout.write('\n') |
| |
| def unpack(wheelfile, dest='.'): |
| """Unpack a wheel. |
| |
| Wheel content will be unpacked to {dest}/{name}-{ver}, where {name} |
| is the package name and {ver} its version. |
| |
| :param wheelfile: The path to the wheel. |
| :param dest: Destination directory (default to current directory). |
| """ |
| wf = WheelFile(wheelfile) |
| namever = wf.parsed_filename.group('namever') |
| destination = os.path.join(dest, namever) |
| sys.stderr.write("Unpacking to: %s\n" % (destination)) |
| wf.zipfile.extractall(destination) |
| wf.zipfile.close() |
| |
| def install(requirements, requirements_file=None, |
| wheel_dirs=None, force=False, list_files=False, |
| dry_run=False): |
| """Install wheels. |
| |
| :param requirements: A list of requirements or wheel files to install. |
| :param requirements_file: A file containing requirements to install. |
| :param wheel_dirs: A list of directories to search for wheels. |
| :param force: Install a wheel file even if it is not compatible. |
| :param list_files: Only list the files to install, don't install them. |
| :param dry_run: Do everything but the actual install. |
| """ |
| |
| # If no wheel directories specified, use the WHEELPATH environment |
| # variable, or the current directory if that is not set. |
| if not wheel_dirs: |
| wheelpath = os.getenv("WHEELPATH") |
| if wheelpath: |
| wheel_dirs = wheelpath.split(os.pathsep) |
| else: |
| wheel_dirs = [ os.path.curdir ] |
| |
| # Get a list of all valid wheels in wheel_dirs |
| all_wheels = [] |
| for d in wheel_dirs: |
| for w in os.listdir(d): |
| if w.endswith('.whl'): |
| wf = WheelFile(os.path.join(d, w)) |
| if wf.compatible: |
| all_wheels.append(wf) |
| |
| # If there is a requirements file, add it to the list of requirements |
| if requirements_file: |
| # If the file doesn't exist, search for it in wheel_dirs |
| # This allows standard requirements files to be stored with the |
| # wheels. |
| if not os.path.exists(requirements_file): |
| for d in wheel_dirs: |
| name = os.path.join(d, requirements_file) |
| if os.path.exists(name): |
| requirements_file = name |
| break |
| |
| with open(requirements_file) as fd: |
| requirements.extend(fd) |
| |
| to_install = [] |
| for req in requirements: |
| if req.endswith('.whl'): |
| # Explicitly specified wheel filename |
| if os.path.exists(req): |
| wf = WheelFile(req) |
| if wf.compatible or force: |
| to_install.append(wf) |
| else: |
| msg = ("{0} is not compatible with this Python. " |
| "--force to install anyway.".format(req)) |
| raise WheelError(msg) |
| else: |
| # We could search on wheel_dirs, but it's probably OK to |
| # assume the user has made an error. |
| raise WheelError("No such wheel file: {}".format(req)) |
| continue |
| |
| # We have a requirement spec |
| # If we don't have pkg_resources, this will raise an exception |
| matches = matches_requirement(req, all_wheels) |
| if not matches: |
| raise WheelError("No match for requirement {}".format(req)) |
| to_install.append(max(matches)) |
| |
| # We now have a list of wheels to install |
| if list_files: |
| sys.stdout.write("Installing:\n") |
| |
| if dry_run: |
| return |
| |
| for wf in to_install: |
| if list_files: |
| sys.stdout.write(" {0}\n".format(wf.filename)) |
| continue |
| wf.install(force=force) |
| wf.zipfile.close() |
| |
| def install_scripts(distributions): |
| """ |
| Regenerate the entry_points console_scripts for the named distribution. |
| """ |
| try: |
| from setuptools.command import easy_install |
| import pkg_resources |
| except ImportError: |
| raise RuntimeError("'wheel install_scripts' needs setuptools.") |
| |
| for dist in distributions: |
| pkg_resources_dist = pkg_resources.get_distribution(dist) |
| install = get_install_command(dist) |
| command = easy_install.easy_install(install.distribution) |
| command.args = ['wheel'] # dummy argument |
| command.finalize_options() |
| command.install_egg_scripts(pkg_resources_dist) |
| |
| def convert(installers, dest_dir, verbose): |
| require_pkgresources('wheel convert') |
| |
| # Only support wheel convert if pkg_resources is present |
| from ..wininst2wheel import bdist_wininst2wheel |
| from ..egg2wheel import egg2wheel |
| |
| for pat in installers: |
| for installer in iglob(pat): |
| if os.path.splitext(installer)[1] == '.egg': |
| conv = egg2wheel |
| else: |
| conv = bdist_wininst2wheel |
| if verbose: |
| sys.stdout.write("{0}... ".format(installer)) |
| sys.stdout.flush() |
| conv(installer, dest_dir) |
| if verbose: |
| sys.stdout.write("OK\n") |
| |
| def parser(): |
| p = argparse.ArgumentParser() |
| s = p.add_subparsers(help="commands") |
| |
| def keygen_f(args): |
| keygen() |
| keygen_parser = s.add_parser('keygen', help='Generate signing key') |
| keygen_parser.set_defaults(func=keygen_f) |
| |
| def sign_f(args): |
| sign(args.wheelfile) |
| sign_parser = s.add_parser('sign', help='Sign wheel') |
| sign_parser.add_argument('wheelfile', help='Wheel file') |
| sign_parser.set_defaults(func=sign_f) |
| |
| def unsign_f(args): |
| unsign(args.wheelfile) |
| unsign_parser = s.add_parser('unsign', help=unsign.__doc__) |
| unsign_parser.add_argument('wheelfile', help='Wheel file') |
| unsign_parser.set_defaults(func=unsign_f) |
| |
| def verify_f(args): |
| verify(args.wheelfile) |
| verify_parser = s.add_parser('verify', help=verify.__doc__) |
| verify_parser.add_argument('wheelfile', help='Wheel file') |
| verify_parser.set_defaults(func=verify_f) |
| |
| def unpack_f(args): |
| unpack(args.wheelfile, args.dest) |
| unpack_parser = s.add_parser('unpack', help='Unpack wheel') |
| unpack_parser.add_argument('--dest', '-d', help='Destination directory', |
| default='.') |
| unpack_parser.add_argument('wheelfile', help='Wheel file') |
| unpack_parser.set_defaults(func=unpack_f) |
| |
| def install_f(args): |
| install(args.requirements, args.requirements_file, |
| args.wheel_dirs, args.force, args.list_files) |
| install_parser = s.add_parser('install', help='Install wheels') |
| install_parser.add_argument('requirements', nargs='*', |
| help='Requirements to install.') |
| install_parser.add_argument('--force', default=False, |
| action='store_true', |
| help='Install incompatible wheel files.') |
| install_parser.add_argument('--wheel-dir', '-d', action='append', |
| dest='wheel_dirs', |
| help='Directories containing wheels.') |
| install_parser.add_argument('--requirements-file', '-r', |
| help="A file containing requirements to " |
| "install.") |
| install_parser.add_argument('--list', '-l', default=False, |
| dest='list_files', |
| action='store_true', |
| help="List wheels which would be installed, " |
| "but don't actually install anything.") |
| install_parser.set_defaults(func=install_f) |
| |
| def install_scripts_f(args): |
| install_scripts(args.distributions) |
| install_scripts_parser = s.add_parser('install-scripts', help='Install console_scripts') |
| install_scripts_parser.add_argument('distributions', nargs='*', |
| help='Regenerate console_scripts for these distributions') |
| install_scripts_parser.set_defaults(func=install_scripts_f) |
| |
| def convert_f(args): |
| convert(args.installers, args.dest_dir, args.verbose) |
| convert_parser = s.add_parser('convert', help='Convert egg or wininst to wheel') |
| convert_parser.add_argument('installers', nargs='*', help='Installers to convert') |
| convert_parser.add_argument('--dest-dir', '-d', default=os.path.curdir, |
| help="Directory to store wheels (default %(default)s)") |
| convert_parser.add_argument('--verbose', '-v', action='store_true') |
| convert_parser.set_defaults(func=convert_f) |
| |
| def version_f(args): |
| from .. import __version__ |
| sys.stdout.write("wheel %s\n" % __version__) |
| version_parser = s.add_parser('version', help='Print version and exit') |
| version_parser.set_defaults(func=version_f) |
| |
| def help_f(args): |
| p.print_help() |
| help_parser = s.add_parser('help', help='Show this help') |
| help_parser.set_defaults(func=help_f) |
| |
| return p |
| |
| def main(): |
| p = parser() |
| args = p.parse_args() |
| if not hasattr(args, 'func'): |
| p.print_help() |
| else: |
| # XXX on Python 3.3 we get 'args has no func' rather than short help. |
| try: |
| args.func(args) |
| return 0 |
| except WheelError as e: |
| sys.stderr.write(e.message + "\n") |
| return 1 |
