git.scottworley.com Git - pinch/blob - pinch.py

   1 import argparse
   2 import configparser
   3 import filecmp
   4 import functools
   5 import getpass
   6 import hashlib
   7 import operator
   8 import os
   9 import os.path
  10 import shlex
  11 import shutil
  12 import subprocess
  13 import sys
  14 import tarfile
  15 import tempfile
  16 import types
  17 import urllib.parse
  18 import urllib.request
  19 import xml.dom.minidom
  20
  21 from typing import (
  22     Callable,
  23     Dict,
  24     Iterable,
  25     List,
  26     Mapping,
  27     NamedTuple,
  28     NewType,
  29     Optional,
  30     Set,
  31     Tuple,
  32     Type,
  33     TypeVar,
  34     Union,
  35 )
  36
  37 import git_cache
  38
  39 # Use xdg module when it's less painful to have as a dependency
  40
  41
  42 class XDG(NamedTuple):
  43     XDG_CACHE_HOME: str
  44
  45
  46 xdg = XDG(
  47     XDG_CACHE_HOME=os.getenv(
  48         'XDG_CACHE_HOME',
  49         os.path.expanduser('~/.cache')))
  50
  51
  52 class VerificationError(Exception):
  53     pass
  54
  55
  56 class Verification:
  57
  58     def __init__(self) -> None:
  59         self.line_length = 0
  60
  61     def status(self, s: str) -> None:
  62         print(s, end=' ', file=sys.stderr, flush=True)
  63         self.line_length += 1 + len(s)  # Unicode??
  64
  65     @staticmethod
  66     def _color(s: str, c: int) -> str:
  67         return f'\033[{c:2d}m{s}\033[00m'
  68
  69     def result(self, r: bool) -> None:
  70         message, color = {True: ('OK ', 92), False: ('FAIL', 91)}[r]
  71         length = len(message)
  72         cols = shutil.get_terminal_size().columns or 80
  73         pad = (cols - (self.line_length + length)) % cols
  74         print(' ' * pad + self._color(message, color), file=sys.stderr)
  75         self.line_length = 0
  76         if not r:
  77             raise VerificationError()
  78
  79     def check(self, s: str, r: bool) -> None:
  80         self.status(s)
  81         self.result(r)
  82
  83     def ok(self) -> None:
  84         self.result(True)
  85
  86
  87 Digest16 = NewType('Digest16', str)
  88 Digest32 = NewType('Digest32', str)
  89
  90
  91 class ChannelTableEntry(types.SimpleNamespace):
  92     absolute_url: str
  93     digest: Digest16
  94     file: str
  95     size: int
  96     url: str
  97
  98
  99 class AliasPin(NamedTuple):
 100     pass
 101
 102
 103 class SymlinkPin(NamedTuple):
 104     @property
 105     def release_name(self) -> str:
 106         return 'link'
 107
 108
 109 class GitPin(NamedTuple):
 110     git_revision: str
 111     release_name: str
 112
 113
 114 class ChannelPin(NamedTuple):
 115     git_revision: str
 116     release_name: str
 117     tarball_url: str
 118     tarball_sha256: str
 119
 120
 121 Pin = Union[AliasPin, SymlinkPin, GitPin, ChannelPin]
 122
 123
 124 def copy_to_nix_store(v: Verification, filename: str) -> str:
 125     v.status('Putting tarball in Nix store')
 126     process = subprocess.run(
 127         ['nix-store', '--add', filename], stdout=subprocess.PIPE)
 128     v.result(process.returncode == 0)
 129     return process.stdout.decode().strip()  # type: ignore  # (for old mypy)
 130
 131
 132 def symlink_archive(v: Verification, path: str) -> str:
 133     with tempfile.TemporaryDirectory() as td:
 134         archive_filename = os.path.join(td, 'link.tar.gz')
 135         os.symlink(path, os.path.join(td, 'link'))
 136         with tarfile.open(archive_filename, mode='x:gz') as t:
 137             t.add(os.path.join(td, 'link'), arcname='link')
 138         return copy_to_nix_store(v, archive_filename)
 139
 140
 141 class AliasSearchPath(NamedTuple):
 142     alias_of: str
 143
 144     # pylint: disable=no-self-use
 145     def pin(self, _: Verification, __: Optional[Pin]) -> AliasPin:
 146         return AliasPin()
 147
 148
 149 class SymlinkSearchPath(NamedTuple):
 150     path: str
 151
 152     # pylint: disable=no-self-use
 153     def pin(self, _: Verification, __: Optional[Pin]) -> SymlinkPin:
 154         return SymlinkPin()
 155
 156     def fetch(self, v: Verification, _: Pin) -> str:
 157         return symlink_archive(v, self.path)
 158
 159
 160 class GitSearchPath(NamedTuple):
 161     git_ref: str
 162     git_repo: str
 163
 164     def pin(self, v: Verification, old_pin: Optional[Pin]) -> GitPin:
 165         _, new_revision = git_cache.fetch(self.git_repo, self.git_ref)
 166         if old_pin is not None:
 167             assert isinstance(old_pin, GitPin)
 168             verify_git_ancestry(v, self, old_pin.git_revision, new_revision)
 169         return GitPin(release_name=git_revision_name(v, self, new_revision),
 170                       git_revision=new_revision)
 171
 172     def fetch(self, v: Verification, pin: Pin) -> str:
 173         assert isinstance(pin, GitPin)
 174         git_cache.ensure_rev_available(
 175             self.git_repo, self.git_ref, pin.git_revision)
 176         return git_get_tarball(v, self, pin)
 177
 178
 179 class ChannelSearchPath(NamedTuple):
 180     channel_url: str
 181     git_ref: str
 182     git_repo: str
 183
 184     def pin(self, v: Verification, old_pin: Optional[Pin]) -> ChannelPin:
 185         if old_pin is not None:
 186             assert isinstance(old_pin, ChannelPin)
 187
 188         channel_html, forwarded_url = fetch_channel(v, self)
 189         table, new_gitpin = parse_channel(v, channel_html)
 190         if old_pin is not None and old_pin.git_revision == new_gitpin.git_revision:
 191             return old_pin
 192         fetch_resources(v, new_gitpin, forwarded_url, table)
 193         git_cache.ensure_rev_available(
 194             self.git_repo, self.git_ref, new_gitpin.git_revision)
 195         if old_pin is not None:
 196             verify_git_ancestry(
 197                 v, self, old_pin.git_revision, new_gitpin.git_revision)
 198         check_channel_contents(v, self, table, new_gitpin)
 199         return ChannelPin(
 200             release_name=new_gitpin.release_name,
 201             tarball_url=table['nixexprs.tar.xz'].absolute_url,
 202             tarball_sha256=table['nixexprs.tar.xz'].digest,
 203             git_revision=new_gitpin.git_revision)
 204
 205     # pylint: disable=no-self-use
 206     def fetch(self, v: Verification, pin: Pin) -> str:
 207         assert isinstance(pin, ChannelPin)
 208
 209         return fetch_with_nix_prefetch_url(
 210             v, pin.tarball_url, Digest16(pin.tarball_sha256))
 211
 212
 213 SearchPath = Union[AliasSearchPath,
 214                    SymlinkSearchPath,
 215                    GitSearchPath,
 216                    ChannelSearchPath]
 217 TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
 218
 219
 220 def compare(a: str, b: str) -> Tuple[List[str], List[str], List[str]]:
 221
 222     def throw(error: OSError) -> None:
 223         raise error
 224
 225     def join(x: str, y: str) -> str:
 226         return y if x == '.' else os.path.join(x, y)
 227
 228     def recursive_files(d: str) -> Iterable[str]:
 229         all_files: List[str] = []
 230         for path, dirs, files in os.walk(d, onerror=throw):
 231             rel = os.path.relpath(path, start=d)
 232             all_files.extend(join(rel, f) for f in files)
 233             for dir_or_link in dirs:
 234                 if os.path.islink(join(path, dir_or_link)):
 235                     all_files.append(join(rel, dir_or_link))
 236         return all_files
 237
 238     def exclude_dot_git(files: Iterable[str]) -> Iterable[str]:
 239         return (f for f in files if not f.startswith('.git/'))
 240
 241     files = functools.reduce(
 242         operator.or_, (set(
 243             exclude_dot_git(
 244                 recursive_files(x))) for x in [a, b]))
 245     return filecmp.cmpfiles(a, b, files, shallow=False)
 246
 247
 248 def fetch_channel(
 249         v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
 250     v.status(f'Fetching channel from {channel.channel_url}')
 251     with urllib.request.urlopen(channel.channel_url, timeout=10) as request:
 252         channel_html = request.read().decode()
 253         forwarded_url = request.geturl()
 254         v.result(request.status == 200)
 255     v.check('Got forwarded', channel.channel_url != forwarded_url)
 256     return channel_html, forwarded_url
 257
 258
 259 def parse_channel(v: Verification, channel_html: str) \
 260         -> Tuple[Dict[str, ChannelTableEntry], GitPin]:
 261     v.status('Parsing channel description as XML')
 262     d = xml.dom.minidom.parseString(channel_html)
 263     v.ok()
 264
 265     v.status('Extracting release name:')
 266     title_name = d.getElementsByTagName(
 267         'title')[0].firstChild.nodeValue.split()[2]
 268     h1_name = d.getElementsByTagName('h1')[0].firstChild.nodeValue.split()[2]
 269     v.status(title_name)
 270     v.result(title_name == h1_name)
 271
 272     v.status('Extracting git commit:')
 273     git_commit_node = d.getElementsByTagName('tt')[0]
 274     git_revision = git_commit_node.firstChild.nodeValue
 275     v.status(git_revision)
 276     v.ok()
 277     v.status('Verifying git commit label')
 278     v.result(git_commit_node.previousSibling.nodeValue == 'Git commit ')
 279
 280     v.status('Parsing table')
 281     table: Dict[str, ChannelTableEntry] = {}
 282     for row in d.getElementsByTagName('tr')[1:]:
 283         name = row.childNodes[0].firstChild.firstChild.nodeValue
 284         url = row.childNodes[0].firstChild.getAttribute('href')
 285         size = int(row.childNodes[1].firstChild.nodeValue)
 286         digest = Digest16(row.childNodes[2].firstChild.firstChild.nodeValue)
 287         table[name] = ChannelTableEntry(url=url, digest=digest, size=size)
 288     v.ok()
 289     return table, GitPin(release_name=title_name, git_revision=git_revision)
 290
 291
 292 def digest_string(s: bytes) -> Digest16:
 293     return Digest16(hashlib.sha256(s).hexdigest())
 294
 295
 296 def digest_file(filename: str) -> Digest16:
 297     hasher = hashlib.sha256()
 298     with open(filename, 'rb') as f:
 299         # pylint: disable=cell-var-from-loop
 300         for block in iter(lambda: f.read(4096), b''):
 301             hasher.update(block)
 302     return Digest16(hasher.hexdigest())
 303
 304
 305 @functools.lru_cache
 306 def _experimental_flag_needed(v: Verification) -> bool:
 307     v.status('Checking Nix version')
 308     process = subprocess.run(['nix', '--help'], stdout=subprocess.PIPE)
 309     v.result(process.returncode == 0)
 310     return b'--experimental-features' in process.stdout
 311
 312
 313 def _nix_command(v: Verification) -> List[str]:
 314     return ['nix', '--experimental-features',
 315             'nix-command'] if _experimental_flag_needed(v) else ['nix']
 316
 317
 318 def to_Digest16(v: Verification, digest32: Digest32) -> Digest16:
 319     v.status('Converting digest to base16')
 320     process = subprocess.run(_nix_command(v) + [
 321         'to-base16',
 322         '--type',
 323         'sha256',
 324         digest32],
 325         stdout=subprocess.PIPE)
 326     v.result(process.returncode == 0)
 327     return Digest16(process.stdout.decode().strip())
 328
 329
 330 def to_Digest32(v: Verification, digest16: Digest16) -> Digest32:
 331     v.status('Converting digest to base32')
 332     process = subprocess.run(_nix_command(v) + [
 333         'to-base32',
 334         '--type',
 335         'sha256',
 336         digest16],
 337         stdout=subprocess.PIPE)
 338     v.result(process.returncode == 0)
 339     return Digest32(process.stdout.decode().strip())
 340
 341
 342 def fetch_with_nix_prefetch_url(
 343         v: Verification,
 344         url: str,
 345         digest: Digest16) -> str:
 346     v.status(f'Fetching {url}')
 347     process = subprocess.run(
 348         ['nix-prefetch-url', '--print-path', url, digest], stdout=subprocess.PIPE)
 349     v.result(process.returncode == 0)
 350     prefetch_digest, path, empty = process.stdout.decode().split('\n')
 351     assert empty == ''
 352     v.check("Verifying nix-prefetch-url's digest",
 353             to_Digest16(v, Digest32(prefetch_digest)) == digest)
 354     v.status(f"Verifying digest of {path}")
 355     file_digest = digest_file(path)
 356     v.result(file_digest == digest)
 357     return path  # type: ignore  # (for old mypy)
 358
 359
 360 def fetch_resources(
 361         v: Verification,
 362         pin: GitPin,
 363         forwarded_url: str,
 364         table: Dict[str, ChannelTableEntry]) -> None:
 365     for resource in ['git-revision', 'nixexprs.tar.xz']:
 366         fields = table[resource]
 367         fields.absolute_url = urllib.parse.urljoin(forwarded_url, fields.url)
 368         fields.file = fetch_with_nix_prefetch_url(
 369             v, fields.absolute_url, fields.digest)
 370     v.status('Verifying git commit on main page matches git commit in table')
 371     with open(table['git-revision'].file, encoding='utf-8') as rev_file:
 372         v.result(rev_file.read(999) == pin.git_revision)
 373
 374
 375 def tarball_cache_file(channel: TarrableSearchPath, pin: GitPin) -> str:
 376     return os.path.join(
 377         xdg.XDG_CACHE_HOME,
 378         'pinch/git-tarball',
 379         f'{digest_string(channel.git_repo.encode())}-{pin.git_revision}-{pin.release_name}')
 380
 381
 382 def verify_git_ancestry(
 383         v: Verification,
 384         channel: TarrableSearchPath,
 385         old_revision: str,
 386         new_revision: str) -> None:
 387     cachedir = git_cache.git_cachedir(channel.git_repo)
 388     v.status(f'Verifying rev is an ancestor of previous rev {old_revision}')
 389     process = subprocess.run(['git',
 390                               '-C',
 391                               cachedir,
 392                               'merge-base',
 393                               '--is-ancestor',
 394                               old_revision,
 395                               new_revision])
 396     v.result(process.returncode == 0)
 397
 398
 399 def compare_tarball_and_git(
 400         v: Verification,
 401         pin: GitPin,
 402         channel_contents: str,
 403         git_contents: str) -> None:
 404     v.status('Comparing channel tarball with git checkout')
 405     match, mismatch, errors = compare(os.path.join(
 406         channel_contents, pin.release_name), git_contents)
 407     v.ok()
 408     v.check(f'{len(match)} files match', len(match) > 0)
 409     v.check(f'{len(mismatch)} files differ', len(mismatch) == 0)
 410     expected_errors = [
 411         '.git-revision',
 412         '.version-suffix',
 413         'nixpkgs',
 414         'programs.sqlite',
 415         'svn-revision']
 416     benign_errors = []
 417     for ee in expected_errors:
 418         if ee in errors:
 419             errors.remove(ee)
 420             benign_errors.append(ee)
 421     v.check(f'{len(errors)} unexpected incomparable files', len(errors) == 0)
 422     v.check(
 423         f'({len(benign_errors)} of {len(expected_errors)} expected incomparable files)',
 424         len(benign_errors) == len(expected_errors))
 425
 426
 427 def extract_tarball(
 428         v: Verification,
 429         table: Dict[str, ChannelTableEntry],
 430         dest: str) -> None:
 431     v.status(f"Extracting tarball {table['nixexprs.tar.xz'].file}")
 432     shutil.unpack_archive(table['nixexprs.tar.xz'].file, dest)
 433     v.ok()
 434
 435
 436 def git_checkout(
 437         v: Verification,
 438         channel: TarrableSearchPath,
 439         pin: GitPin,
 440         dest: str) -> None:
 441     v.status('Checking out corresponding git revision')
 442     with subprocess.Popen(
 443             ['git', '-C', git_cache.git_cachedir(channel.git_repo), 'archive', pin.git_revision],
 444             stdout=subprocess.PIPE) as git:
 445         with subprocess.Popen(['tar', 'x', '-C', dest, '-f', '-'], stdin=git.stdout) as tar:
 446             if git.stdout:
 447                 git.stdout.close()
 448             tar.wait()
 449             git.wait()
 450             v.result(git.returncode == 0 and tar.returncode == 0)
 451
 452
 453 def git_get_tarball(
 454         v: Verification,
 455         channel: TarrableSearchPath,
 456         pin: GitPin) -> str:
 457     cache_file = tarball_cache_file(channel, pin)
 458     if os.path.exists(cache_file):
 459         with open(cache_file, encoding='utf-8') as f:
 460             cached_tarball = f.read(9999)
 461             if os.path.exists(cached_tarball):
 462                 return cached_tarball
 463
 464     with tempfile.TemporaryDirectory() as output_dir:
 465         output_filename = os.path.join(
 466             output_dir, pin.release_name + '.tar.xz')
 467         with open(output_filename, 'w', encoding='utf-8') as output_file:
 468             v.status(f'Generating tarball for git revision {pin.git_revision}')
 469             with subprocess.Popen(
 470                     ['git', '-C', git_cache.git_cachedir(channel.git_repo),
 471                         'archive', f'--prefix={pin.release_name}/', pin.git_revision],
 472                     stdout=subprocess.PIPE) as git:
 473                 with subprocess.Popen(['xz'], stdin=git.stdout, stdout=output_file) as xz:
 474                     xz.wait()
 475                     git.wait()
 476                     v.result(git.returncode == 0 and xz.returncode == 0)
 477
 478         store_tarball = copy_to_nix_store(v, output_filename)
 479
 480         os.makedirs(os.path.dirname(cache_file), exist_ok=True)
 481         with open(cache_file, 'w', encoding='utf-8') as f:
 482             f.write(store_tarball)
 483         return store_tarball  # type: ignore  # (for old mypy)
 484
 485
 486 def check_channel_metadata(
 487         v: Verification,
 488         pin: GitPin,
 489         channel_contents: str) -> None:
 490     v.status('Verifying git commit in channel tarball')
 491     with open(os.path.join(channel_contents, pin.release_name, '.git-revision'),
 492               encoding='utf-8') as f:
 493         v.result(f.read(999) == pin.git_revision)
 494
 495     v.status(
 496         f'Verifying version-suffix is a suffix of release name {pin.release_name}:')
 497     with open(os.path.join(channel_contents, pin.release_name, '.version-suffix'),
 498               encoding='utf-8') as f:
 499         version_suffix = f.read(999)
 500     v.status(version_suffix)
 501     v.result(pin.release_name.endswith(version_suffix))
 502
 503
 504 def check_channel_contents(
 505         v: Verification,
 506         channel: TarrableSearchPath,
 507         table: Dict[str, ChannelTableEntry],
 508         pin: GitPin) -> None:
 509     with tempfile.TemporaryDirectory() as channel_contents, \
 510             tempfile.TemporaryDirectory() as git_contents:
 511
 512         extract_tarball(v, table, channel_contents)
 513         check_channel_metadata(v, pin, channel_contents)
 514
 515         git_checkout(v, channel, pin, git_contents)
 516
 517         compare_tarball_and_git(v, pin, channel_contents, git_contents)
 518
 519         v.status('Removing temporary directories')
 520     v.ok()
 521
 522
 523 def git_revision_name(
 524         v: Verification,
 525         channel: TarrableSearchPath,
 526         git_revision: str) -> str:
 527     v.status('Getting commit date')
 528     process = subprocess.run(['git',
 529                               '-C',
 530                               git_cache.git_cachedir(channel.git_repo),
 531                               'log',
 532                               '-n1',
 533                               '--format=%ct-%h',
 534                               '--abbrev=11',
 535                               '--no-show-signature',
 536                               git_revision],
 537                              stdout=subprocess.PIPE)
 538     v.result(process.returncode == 0 and process.stdout != b'')
 539     return f'{os.path.basename(channel.git_repo)}-{process.stdout.decode().strip()}'
 540
 541
 542 K = TypeVar('K')
 543 V = TypeVar('V')
 544
 545
 546 def partition_dict(pred: Callable[[K, V], bool],
 547                    d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]:
 548     selected: Dict[K, V] = {}
 549     remaining: Dict[K, V] = {}
 550     for k, v in d.items():
 551         if pred(k, v):
 552             selected[k] = v
 553         else:
 554             remaining[k] = v
 555     return selected, remaining
 556
 557
 558 def filter_dict(d: Dict[K, V], fields: Set[K]
 559                 ) -> Tuple[Dict[K, V], Dict[K, V]]:
 560     return partition_dict(lambda k, v: k in fields, d)
 561
 562
 563 def read_config_section(
 564         conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]:
 565     mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = {
 566         'alias': (AliasSearchPath, AliasPin),
 567         'channel': (ChannelSearchPath, ChannelPin),
 568         'git': (GitSearchPath, GitPin),
 569         'symlink': (SymlinkSearchPath, SymlinkPin),
 570     }
 571     SP, P = mapping[conf['type']]
 572     _, all_fields = filter_dict(dict(conf.items()), set(['type']))
 573     pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
 574     # Error suppression works around https://github.com/python/mypy/issues/9007
 575     pin_present = pin_fields or P._fields == ()
 576     pin = P(**pin_fields) if pin_present else None  # type: ignore
 577     return SP(**remaining_fields), pin
 578
 579
 580 def read_pinned_config_section(
 581         section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
 582     sp, pin = read_config_section(conf)
 583     if pin is None:
 584         raise Exception(
 585             f'Cannot update unpinned channel "{section}" (Run "pin" before "update")')
 586     return sp, pin
 587
 588
 589 def read_config(filename: str) -> configparser.ConfigParser:
 590     config = configparser.ConfigParser()
 591     with open(filename, encoding='utf-8') as f:
 592         config.read_file(f, filename)
 593     return config
 594
 595
 596 def read_config_files(
 597         filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]:
 598     merged_config: Dict[str, configparser.SectionProxy] = {}
 599     for file in filenames:
 600         config = read_config(file)
 601         for section in config.sections():
 602             if section in merged_config:
 603                 raise Exception('Duplicate channel "{section}"')
 604             merged_config[section] = config[section]
 605     return merged_config
 606
 607
 608 def pinCommand(args: argparse.Namespace) -> None:
 609     v = Verification()
 610     config = read_config(args.channels_file)
 611     for section in config.sections():
 612         if args.channels and section not in args.channels:
 613             continue
 614
 615         sp, old_pin = read_config_section(config[section])
 616
 617         config[section].update(sp.pin(v, old_pin)._asdict())
 618
 619     with open(args.channels_file, 'w', encoding='utf-8') as configfile:
 620         config.write(configfile)
 621
 622
 623 def updateCommand(args: argparse.Namespace) -> None:
 624     v = Verification()
 625     exprs: Dict[str, str] = {}
 626     profile_manifest = os.path.join(args.profile, "manifest.nix")
 627     search_paths: List[str] = [
 628         "-I", "pinch_profile=" + args.profile,
 629         "-I", "pinch_profile_manifest=" + os.readlink(profile_manifest)
 630     ] if os.path.exists(profile_manifest) else []
 631     config = {
 632         section: read_pinned_config_section(section, conf) for section,
 633         conf in read_config_files(
 634             args.channels_file).items()}
 635     alias, nonalias = partition_dict(
 636         lambda k, v: isinstance(v[0], AliasSearchPath), config)
 637
 638     for section, (sp, pin) in sorted(nonalias.items()):
 639         assert not isinstance(sp, AliasSearchPath)  # mypy can't see through
 640         assert not isinstance(pin, AliasPin)        # partition_dict()
 641         tarball = sp.fetch(v, pin)
 642         search_paths.extend(
 643             ["-I", f"pinch_tarball_for_{pin.release_name}={tarball}"])
 644         exprs[section] = (
 645             f'f: f {{ name = "{pin.release_name}"; channelName = "%s"; '
 646             f'src = builtins.storePath "{tarball}"; }}')
 647
 648     for section, (sp, pin) in alias.items():
 649         assert isinstance(sp, AliasSearchPath)  # For mypy
 650         exprs[section] = exprs[sp.alias_of]
 651
 652     command = [
 653         'nix-env',
 654         '--profile',
 655         args.profile,
 656         '--show-trace',
 657         '--file',
 658         '<nix/unpack-channel.nix>',
 659         '--install',
 660         '--remove-all',
 661     ] + search_paths + ['--from-expression'] + [
 662         exprs[name] % name for name in sorted(exprs.keys())]
 663     if args.dry_run:
 664         print(' '.join(map(shlex.quote, command)))
 665     else:
 666         v.status('Installing channels with nix-env')
 667         process = subprocess.run(command)
 668         v.result(process.returncode == 0)
 669
 670
 671 def main() -> None:
 672     parser = argparse.ArgumentParser(prog='pinch')
 673     subparsers = parser.add_subparsers(dest='mode', required=True)
 674     parser_pin = subparsers.add_parser('pin')
 675     parser_pin.add_argument('channels_file', type=str)
 676     parser_pin.add_argument('channels', type=str, nargs='*')
 677     parser_pin.set_defaults(func=pinCommand)
 678     parser_update = subparsers.add_parser('update')
 679     parser_update.add_argument('--dry-run', action='store_true')
 680     parser_update.add_argument('--profile', default=(
 681         f'/nix/var/nix/profiles/per-user/{getpass.getuser()}/channels'))
 682     parser_update.add_argument('channels_file', type=str, nargs='+')
 683     parser_update.set_defaults(func=updateCommand)
 684     args = parser.parse_args()
 685     args.func(args)
 686
 687
 688 if __name__ == '__main__':
 689     main()