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

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