]> git.scottworley.com Git - pinch/blob - pinch.py
c26070b17ee00eac25b191454f32f54b45e58b3c
[pinch] / 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 '\033[%2dm%s\033[00m' % (c, s)
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('Fetching channel')
251 request = urllib.request.urlopen(channel.channel_url, timeout=10)
252 channel_html = request.read().decode()
253 forwarded_url = request.geturl()
254 v.result(request.status == 200) # type: ignore # (for old mypy)
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 _experimental_flag_needed = None
306
307
308 def _nix_command(v: Verification) -> List[str]:
309 global _experimental_flag_needed
310 if _experimental_flag_needed is None:
311 v.status('Checking Nix version')
312 process = subprocess.run(['nix', '--help'], stdout=subprocess.PIPE)
313 v.result(process.returncode == 0)
314 _experimental_flag_needed = b'--experimental-features' in process.stdout
315 return ['nix', '--experimental-features',
316 'nix-command'] if _experimental_flag_needed else ['nix']
317
318
319 def to_Digest16(v: Verification, digest32: Digest32) -> Digest16:
320 v.status('Converting digest to base16')
321 process = subprocess.run(_nix_command(v) + [
322 'to-base16',
323 '--type',
324 'sha256',
325 digest32],
326 stdout=subprocess.PIPE)
327 v.result(process.returncode == 0)
328 return Digest16(process.stdout.decode().strip())
329
330
331 def to_Digest32(v: Verification, digest16: Digest16) -> Digest32:
332 v.status('Converting digest to base32')
333 process = subprocess.run(_nix_command(v) + [
334 'to-base32',
335 '--type',
336 'sha256',
337 digest16],
338 stdout=subprocess.PIPE)
339 v.result(process.returncode == 0)
340 return Digest32(process.stdout.decode().strip())
341
342
343 def fetch_with_nix_prefetch_url(
344 v: Verification,
345 url: str,
346 digest: Digest16) -> str:
347 v.status('Fetching %s' % url)
348 process = subprocess.run(
349 ['nix-prefetch-url', '--print-path', url, digest], stdout=subprocess.PIPE)
350 v.result(process.returncode == 0)
351 prefetch_digest, path, empty = process.stdout.decode().split('\n')
352 assert empty == ''
353 v.check("Verifying nix-prefetch-url's digest",
354 to_Digest16(v, Digest32(prefetch_digest)) == digest)
355 v.status("Verifying file digest")
356 file_digest = digest_file(path)
357 v.result(file_digest == digest)
358 return path # type: ignore # (for old mypy)
359
360
361 def fetch_resources(
362 v: Verification,
363 pin: GitPin,
364 forwarded_url: str,
365 table: Dict[str, ChannelTableEntry]) -> None:
366 for resource in ['git-revision', 'nixexprs.tar.xz']:
367 fields = table[resource]
368 fields.absolute_url = urllib.parse.urljoin(forwarded_url, fields.url)
369 fields.file = fetch_with_nix_prefetch_url(
370 v, fields.absolute_url, fields.digest)
371 v.status('Verifying git commit on main page matches git commit in table')
372 v.result(open(table['git-revision'].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 '%s-%s-%s' %
380 (digest_string(channel.git_repo.encode()),
381 pin.git_revision,
382 pin.release_name))
383
384
385 def verify_git_ancestry(
386 v: Verification,
387 channel: TarrableSearchPath,
388 old_revision: str,
389 new_revision: str) -> None:
390 cachedir = git_cache.git_cachedir(channel.git_repo)
391 v.status('Verifying rev is an ancestor of previous rev %s' % old_revision)
392 process = subprocess.run(['git',
393 '-C',
394 cachedir,
395 'merge-base',
396 '--is-ancestor',
397 old_revision,
398 new_revision])
399 v.result(process.returncode == 0)
400
401
402 def compare_tarball_and_git(
403 v: Verification,
404 pin: GitPin,
405 channel_contents: str,
406 git_contents: str) -> None:
407 v.status('Comparing channel tarball with git checkout')
408 match, mismatch, errors = compare(os.path.join(
409 channel_contents, pin.release_name), git_contents)
410 v.ok()
411 v.check('%d files match' % len(match), len(match) > 0)
412 v.check('%d files differ' % len(mismatch), len(mismatch) == 0)
413 expected_errors = [
414 '.git-revision',
415 '.version-suffix',
416 'nixpkgs',
417 'programs.sqlite',
418 'svn-revision']
419 benign_errors = []
420 for ee in expected_errors:
421 if ee in errors:
422 errors.remove(ee)
423 benign_errors.append(ee)
424 v.check(
425 '%d unexpected incomparable files' %
426 len(errors),
427 len(errors) == 0)
428 v.check(
429 '(%d of %d expected incomparable files)' %
430 (len(benign_errors),
431 len(expected_errors)),
432 len(benign_errors) == len(expected_errors))
433
434
435 def extract_tarball(
436 v: Verification,
437 table: Dict[str, ChannelTableEntry],
438 dest: str) -> None:
439 v.status('Extracting tarball %s' % table['nixexprs.tar.xz'].file)
440 shutil.unpack_archive(table['nixexprs.tar.xz'].file, dest)
441 v.ok()
442
443
444 def git_checkout(
445 v: Verification,
446 channel: TarrableSearchPath,
447 pin: GitPin,
448 dest: str) -> None:
449 v.status('Checking out corresponding git revision')
450 git = subprocess.Popen(['git',
451 '-C',
452 git_cache.git_cachedir(channel.git_repo),
453 'archive',
454 pin.git_revision],
455 stdout=subprocess.PIPE)
456 tar = subprocess.Popen(
457 ['tar', 'x', '-C', dest, '-f', '-'], stdin=git.stdout)
458 if git.stdout:
459 git.stdout.close()
460 tar.wait()
461 git.wait()
462 v.result(git.returncode == 0 and tar.returncode == 0)
463
464
465 def git_get_tarball(
466 v: Verification,
467 channel: TarrableSearchPath,
468 pin: GitPin) -> str:
469 cache_file = tarball_cache_file(channel, pin)
470 if os.path.exists(cache_file):
471 cached_tarball = open(cache_file).read(9999)
472 if os.path.exists(cached_tarball):
473 return cached_tarball
474
475 with tempfile.TemporaryDirectory() as output_dir:
476 output_filename = os.path.join(
477 output_dir, pin.release_name + '.tar.xz')
478 with open(output_filename, 'w') as output_file:
479 v.status(
480 'Generating tarball for git revision %s' %
481 pin.git_revision)
482 git = subprocess.Popen(['git',
483 '-C',
484 git_cache.git_cachedir(channel.git_repo),
485 'archive',
486 '--prefix=%s/' % pin.release_name,
487 pin.git_revision],
488 stdout=subprocess.PIPE)
489 xz = subprocess.Popen(['xz'], stdin=git.stdout, stdout=output_file)
490 xz.wait()
491 git.wait()
492 v.result(git.returncode == 0 and xz.returncode == 0)
493
494 store_tarball = copy_to_nix_store(v, output_filename)
495
496 os.makedirs(os.path.dirname(cache_file), exist_ok=True)
497 open(cache_file, 'w').write(store_tarball)
498 return store_tarball # type: ignore # (for old mypy)
499
500
501 def check_channel_metadata(
502 v: Verification,
503 pin: GitPin,
504 channel_contents: str) -> None:
505 v.status('Verifying git commit in channel tarball')
506 v.result(
507 open(
508 os.path.join(
509 channel_contents,
510 pin.release_name,
511 '.git-revision')).read(999) == pin.git_revision)
512
513 v.status(
514 'Verifying version-suffix is a suffix of release name %s:' %
515 pin.release_name)
516 version_suffix = open(
517 os.path.join(
518 channel_contents,
519 pin.release_name,
520 '.version-suffix')).read(999)
521 v.status(version_suffix)
522 v.result(pin.release_name.endswith(version_suffix))
523
524
525 def check_channel_contents(
526 v: Verification,
527 channel: TarrableSearchPath,
528 table: Dict[str, ChannelTableEntry],
529 pin: GitPin) -> None:
530 with tempfile.TemporaryDirectory() as channel_contents, \
531 tempfile.TemporaryDirectory() as git_contents:
532
533 extract_tarball(v, table, channel_contents)
534 check_channel_metadata(v, pin, channel_contents)
535
536 git_checkout(v, channel, pin, git_contents)
537
538 compare_tarball_and_git(v, pin, channel_contents, git_contents)
539
540 v.status('Removing temporary directories')
541 v.ok()
542
543
544 def git_revision_name(
545 v: Verification,
546 channel: TarrableSearchPath,
547 git_revision: str) -> str:
548 v.status('Getting commit date')
549 process = subprocess.run(['git',
550 '-C',
551 git_cache.git_cachedir(channel.git_repo),
552 'log',
553 '-n1',
554 '--format=%ct-%h',
555 '--abbrev=11',
556 '--no-show-signature',
557 git_revision],
558 stdout=subprocess.PIPE)
559 v.result(process.returncode == 0 and process.stdout != b'')
560 return '%s-%s' % (os.path.basename(channel.git_repo),
561 process.stdout.decode().strip())
562
563
564 K = TypeVar('K')
565 V = TypeVar('V')
566
567
568 def partition_dict(pred: Callable[[K, V], bool],
569 d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]:
570 selected: Dict[K, V] = {}
571 remaining: Dict[K, V] = {}
572 for k, v in d.items():
573 if pred(k, v):
574 selected[k] = v
575 else:
576 remaining[k] = v
577 return selected, remaining
578
579
580 def filter_dict(d: Dict[K, V], fields: Set[K]
581 ) -> Tuple[Dict[K, V], Dict[K, V]]:
582 return partition_dict(lambda k, v: k in fields, d)
583
584
585 def read_config_section(
586 conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]:
587 mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = {
588 'alias': (AliasSearchPath, AliasPin),
589 'channel': (ChannelSearchPath, ChannelPin),
590 'git': (GitSearchPath, GitPin),
591 'symlink': (SymlinkSearchPath, SymlinkPin),
592 }
593 SP, P = mapping[conf['type']]
594 _, all_fields = filter_dict(dict(conf.items()), set(['type']))
595 pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
596 # Error suppression works around https://github.com/python/mypy/issues/9007
597 pin_present = pin_fields != {} or P._fields == ()
598 pin = P(**pin_fields) if pin_present else None # type: ignore
599 return SP(**remaining_fields), pin
600
601
602 def read_pinned_config_section(
603 section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
604 sp, pin = read_config_section(conf)
605 if pin is None:
606 raise Exception(
607 'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
608 section)
609 return sp, pin
610
611
612 def read_config(filename: str) -> configparser.ConfigParser:
613 config = configparser.ConfigParser()
614 config.read_file(open(filename), filename)
615 return config
616
617
618 def read_config_files(
619 filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]:
620 merged_config: Dict[str, configparser.SectionProxy] = {}
621 for file in filenames:
622 config = read_config(file)
623 for section in config.sections():
624 if section in merged_config:
625 raise Exception('Duplicate channel "%s"' % section)
626 merged_config[section] = config[section]
627 return merged_config
628
629
630 def pinCommand(args: argparse.Namespace) -> None:
631 v = Verification()
632 config = read_config(args.channels_file)
633 for section in config.sections():
634 if args.channels and section not in args.channels:
635 continue
636
637 sp, old_pin = read_config_section(config[section])
638
639 config[section].update(sp.pin(v, old_pin)._asdict())
640
641 with open(args.channels_file, 'w') as configfile:
642 config.write(configfile)
643
644
645 def updateCommand(args: argparse.Namespace) -> None:
646 v = Verification()
647 exprs: Dict[str, str] = {}
648 config = {
649 section: read_pinned_config_section(section, conf) for section,
650 conf in read_config_files(
651 args.channels_file).items()}
652 alias, nonalias = partition_dict(
653 lambda k, v: isinstance(v[0], AliasSearchPath), config)
654
655 for section, (sp, pin) in nonalias.items():
656 assert not isinstance(sp, AliasSearchPath) # mypy can't see through
657 assert not isinstance(pin, AliasPin) # partition_dict()
658 tarball = sp.fetch(v, pin)
659 exprs[section] = (
660 'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' %
661 (pin.release_name, tarball))
662
663 for section, (sp, pin) in alias.items():
664 assert isinstance(sp, AliasSearchPath) # For mypy
665 exprs[section] = exprs[sp.alias_of]
666
667 command = [
668 'nix-env',
669 '--profile',
670 args.profile,
671 '--show-trace',
672 '--file',
673 '<nix/unpack-channel.nix>',
674 '--install',
675 '--from-expression'] + [exprs[name] % name for name in sorted(exprs.keys())]
676 if args.dry_run:
677 print(' '.join(map(shlex.quote, command)))
678 else:
679 v.status('Installing channels with nix-env')
680 process = subprocess.run(command)
681 v.result(process.returncode == 0)
682
683
684 def main() -> None:
685 parser = argparse.ArgumentParser(prog='pinch')
686 subparsers = parser.add_subparsers(dest='mode', required=True)
687 parser_pin = subparsers.add_parser('pin')
688 parser_pin.add_argument('channels_file', type=str)
689 parser_pin.add_argument('channels', type=str, nargs='*')
690 parser_pin.set_defaults(func=pinCommand)
691 parser_update = subparsers.add_parser('update')
692 parser_update.add_argument('--dry-run', action='store_true')
693 parser_update.add_argument('--profile', default=(
694 '/nix/var/nix/profiles/per-user/%s/channels' % getpass.getuser()))
695 parser_update.add_argument('channels_file', type=str, nargs='+')
696 parser_update.set_defaults(func=updateCommand)
697 args = parser.parse_args()
698 args.func(args)
699
700
701 if __name__ == '__main__':
702 main()