hippotatlib/ownsource.py

   1 # Automatic source code provision (AGPL compliance)
   2
   3 import os
   4 import sys
   5 import fnmatch
   6 import stat
   7 import subprocess
   8
   9 class SourceShipmentPreparer():
  10   def __init__(s, destdir):
  11     # caller may modify, and should read after calling generate()
  12     s.output_name = 'srcbomb.tar.gz'
  13     # defaults, caller can modify after creation
  14     s.src_filter = s.src_filter_glob
  15     s.src_filter_globs = ['/usr/local/*', '!/usr*', '!/etc/*']
  16     s.src_likeparent = s.src_likeparent_git
  17     s.cwd = os.getcwd()
  18     s.find_rune_base = "find -type f -perm -004 \! -path '*/tmp/*'"
  19     s.excludes = ['*~', '*.bak', '*.tmp', '#*#',
  20                   '[0-9][0-9][0-9][0-9]-src.cpio']
  21     s.rune_shell = ['/bin/bash', '-ec']
  22     s.show_pathnames = True
  23     s.rune_cpio = r'''
  24             set -o pipefail
  25             (
  26              %s
  27              # ^ by default, is find ... -print0
  28             ) | (
  29              cpio -Hustar -o --quiet -0 -R 1000:1000 || \
  30              cpio -Hustar -o --quiet -0
  31             )
  32     '''
  33     s.rune_portmanteau = r'''
  34             outfile=$1; shift
  35             rm -f "$outfile"
  36             GZIP=-9 tar zcf "$outfile" "$@"
  37     '''
  38     s.manifest_name='0000-MANIFEST.txt'
  39     # private
  40     s._destdir = destdir
  41     s._outcounter = 0
  42     s._manifest = []
  43     s._dirmap = { }
  44
  45   def src_filter_glob(s, src): # default s.src_filter
  46     for pat in s.src_filter_globs:
  47       negate = pat.startswith('!')
  48       if negate: pat = pat[1:]
  49       if fnmatch.fnmatch(src, pat):
  50         return not negate
  51     return negate
  52
  53   def src_likeparent_git(s, src):
  54     try:
  55       os.stat(os.path.join(src, '.git/.'))
  56     except FileNotFoundError:
  57       return False
  58     else:
  59       return True
  60
  61   def src_parentfinder(s, src, infol): # callers may monkey-patch away
  62     for deref in (False,True):
  63       xinfo = []
  64
  65       search = src
  66       if deref:
  67         search = os.path.realpath(search)
  68
  69       def ascend():
  70         nonlocal search
  71         xinfo.append(os.path.basename(search))
  72         search = os.path.dirname(search)
  73
  74       try:
  75         stab = os.lstat(search)
  76       except FileNotFoundError:
  77         return
  78       if stat.S_ISREG(stab.st_mode):
  79         ascend()
  80
  81       while not os.path.ismount(search):
  82         if s.src_likeparent(search):
  83           xinfo.reverse()
  84           if len(xinfo): infol.append('want=' + os.path.join(*xinfo))
  85           return search
  86
  87         ascend()
  88
  89     # no .git found anywhere
  90     return src
  91
  92   def src_prenormaliser(s, d, infol): # callers may monkey-patch away
  93     return os.path.join(s.cwd, os.path.abspath(d))
  94
  95   def src_find_rune(s, d):
  96     script = s.find_rune_base
  97     for excl in s.excludes + [s.output_name, s.manifest_name]:
  98       assert("'" not in excl)
  99       script += r" \! -name '%s'" % excl
 100     script += ' -print0'
 101     return script
 102
 103   def manifest_append(s, name, infol):
 104     s._manifest.append((name, ' '.join(infol)))
 105
 106   def new_output_name(s, nametail, infol):
 107     s._outcounter += 1
 108     name = '%04d-%s' % (s._outcounter, nametail)
 109     s.manifest_append(name, infol)
 110     return name
 111
 112   def open_output_fh(s, name, mode):
 113     return open(os.path.join(s._destdir, name), mode)
 114
 115   def mk_from_dir(s, d, infol):
 116     try: name = s._dirmap[d]
 117     except KeyError: pass
 118     else:
 119       s.manifest_append(name, infol)
 120       return
 121
 122     if s.show_pathnames: infol.append(d)
 123     find_rune = s.src_find_rune(d)
 124     total_rune = s.rune_cpio % find_rune
 125
 126     name = s.new_output_name('src.cpio', infol)
 127     s._dirmap[d] = name
 128     fh = s.open_output_fh(name, 'wb')
 129
 130     subprocess.run(s.rune_shell + [total_rune],
 131                    cwd=d,
 132                    stdin=subprocess.DEVNULL,
 133                    stdout=fh,
 134                    restore_signals=True,
 135                    check=True)
 136     fh.close()
 137
 138   def mk_from_src(s, d, infol):
 139     d = s.src_prenormaliser(d, infol)
 140     if not s.src_filter(d): return
 141     d = s.src_parentfinder(d, infol)
 142     s.mk_from_dir(d, infol)
 143
 144   def mk_from_srcs(s, dirs=sys.path):
 145     s.mk_from_src(sys.argv[0], ['argv[0]'])
 146     for d in sys.path:
 147       s.mk_from_src(d, ['sys.path'])
 148
 149   def mk_portmanteau(s):
 150     cmdl = s.rune_shell + [ s.rune_portmanteau, 'x',
 151                             s.output_name, s.manifest_name ]
 152     mfh = s.open_output_fh(s.manifest_name,'w')
 153     for (name, info) in s._manifest:
 154       cmdl.append(name)
 155       print('%s\t%s' % (name,info), file=mfh)
 156     mfh.close()
 157     subprocess.run(cmdl,
 158                    cwd=s._destdir,
 159                    stdin=subprocess.DEVNULL,
 160                    stdout=sys.stderr,
 161                    restore_signals=True,
 162                    check=True)
 163
 164   def generate(s):
 165     s.mk_from_srcs()
 166     s.mk_portmanteau()