hippotatlib/ownsource.py

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