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
  44   def src_filter_glob(s, src): # default s.src_filter
  45     for pat in s.src_filter_globs:
  46       negate = pat.startswith('!')
  47       if negate: pat = pat[1:]
  48       if fnmatch.fnmatch(src, pat):
  49         return not negate
  50     return negate
  51
  52   def src_likeparent_git(s, src):
  53     try:
  54       os.stat(os.path.join(src, '.git/.'))
  55     except FileNotFoundError:
  56       return False
  57     else:
  58       return True
  59
  60   def src_parentfinder(s, src, infol): # callers may monkey-patch away
  61     for deref in (False,True):
  62       xinfo = []
  63
  64       search = src
  65       if deref:
  66         search = os.path.realpath(search)
  67
  68       def ascend():
  69         nonlocal search
  70         xinfo.append(os.path.basename(search))
  71         search = os.path.dirname(search)
  72
  73       try:
  74         stab = os.lstat(search)
  75       except FileNotFoundError:
  76         return
  77       if stat.S_ISREG(stab.st_mode):
  78         ascend()
  79
  80       while not os.path.ismount(search):
  81         if s.src_likeparent(search):
  82           xinfo.reverse()
  83           if len(xinfo): infol.append('want=' + os.path.join(*xinfo))
  84           return search
  85
  86         ascend()
  87
  88     # no .git found anywhere
  89     return src
  90
  91   def src_prenormaliser(s, d, infol): # callers may monkey-patch away
  92     return os.path.join(s.cwd, os.path.abspath(d))
  93
  94   def src_find_rune(s, d):
  95     script = s.find_rune_base
  96     for excl in s.excludes + [s.output_name, s.manifest_name]:
  97       assert("'" not in excl)
  98       script += r" \! -name '%s'" % excl
  99     script += ' -print0'
 100     return script
 101
 102   def new_output_name(s, nametail, infol):
 103     s._outcounter += 1
 104     name = '%04d-%s' % (s._outcounter, nametail)
 105     s._manifest.append((name, ' '.join(infol)))
 106     return name
 107
 108   def new_output_fh(s, nametail, infol):
 109     name = s.new_output_name(nametail, infol)
 110     return s.open_output_fh(name, 'wb')
 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     if s.show_pathnames: infol.append(d)
 117     find_rune = s.src_find_rune(d)
 118     total_rune = s.rune_cpio % find_rune
 119     fh = s.new_output_fh('src.cpio', infol)
 120     subprocess.run(s.rune_shell + [total_rune],
 121                    cwd=d,
 122                    stdin=subprocess.DEVNULL,
 123                    stdout=fh,
 124                    restore_signals=True,
 125                    check=True)
 126     fh.close()
 127
 128   def mk_from_src(s, d, infol):
 129     d = s.src_prenormaliser(d, infol)
 130     if not s.src_filter(d): return
 131     d = s.src_parentfinder(d, infol)
 132     s.mk_from_dir(d, infol)
 133
 134   def mk_from_srcs(s, dirs=sys.path):
 135     s.mk_from_src(sys.argv[0], ['argv[0]'])
 136     for d in sys.path:
 137       s.mk_from_src(d, ['sys.path'])
 138
 139   def mk_portmanteau(s):
 140     cmdl = s.rune_shell + [ s.rune_portmanteau, 'x',
 141                             s.output_name, s.manifest_name ]
 142     mfh = s.open_output_fh(s.manifest_name,'w')
 143     for (name, info) in s._manifest:
 144       cmdl.append(name)
 145       print('%s\t%s' % (name,info), file=mfh)
 146     mfh.close()
 147     subprocess.run(s.rune_shell + cmdl,
 148                    cwd=s._destdir,
 149                    stdin=subprocess.DEVNULL,
 150                    stdout=sys.stderr,
 151                    restore_signals=True,
 152                    check=True)
 153
 154   def generate(s):
 155     s.mk_from_srcs()
 156     s.mk_portmanteau()