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_package_globs = [!'/usr/local/*', '/usr*']
  16     s.src_filter_globs = ['!/etc/*']
  17     s.src_likeparent = s.src_likeparent_git
  18     s.report_from_packages = s.report_from_packages_debian
  19     s.cwd = os.getcwd()
  20     s.find_rune_base = "find -type f -perm -004 \! -path '*/tmp/*'"
  21     s.excludes = ['*~', '*.bak', '*.tmp', '#*#',
  22                   '[0-9][0-9][0-9][0-9]-src.cpio']
  23     s.rune_shell = ['/bin/bash', '-ec']
  24     s.show_pathnames = True
  25     s.rune_cpio = r'''
  26             set -o pipefail
  27             (
  28              %s
  29              # ^ by default, is find ... -print0
  30             ) | (
  31              cpio -Hustar -o --quiet -0 -R 1000:1000 || \
  32              cpio -Hustar -o --quiet -0
  33             )
  34     '''
  35     s.rune_portmanteau = r'''
  36             outfile=$1; shift
  37             rm -f "$outfile"
  38             GZIP=-1 tar zcf "$outfile" "$@"
  39     '''
  40     s.manifest_name='0000-MANIFEST.txt'
  41     # private
  42     s._destdir = destdir
  43     s._outcounter = 0
  44     s._manifest = []
  45     s._dirmap = { }
  46     s._package_files = { } # map filename => infol
  47
  48   def thing_matches_globs(s, thing, globs):
  49     for pat in globs:
  50       negate = pat.startswith('!')
  51       if negate: pat = pat[1:]
  52       if fnmatch.fnmatch(thing, pat):
  53         return not negate
  54     return negate
  55
  56   def src_filter_glob(s, src): # default s.src_filter
  57     return s.thing_matches_globs(s, src, s.src_filter_globs)
  58
  59   def src_likeparent_git(s, src):
  60     try:
  61       os.stat(os.path.join(src, '.git/.'))
  62     except FileNotFoundError:
  63       return False
  64     else:
  65       return True
  66
  67   def src_parentfinder(s, src, infol): # callers may monkey-patch away
  68     for deref in (False,True):
  69       xinfo = []
  70
  71       search = src
  72       if deref:
  73         search = os.path.realpath(search)
  74
  75       def ascend():
  76         nonlocal search
  77         xinfo.append(os.path.basename(search))
  78         search = os.path.dirname(search)
  79
  80       try:
  81         stab = os.lstat(search)
  82       except FileNotFoundError:
  83         return
  84       if stat.S_ISREG(stab.st_mode):
  85         ascend()
  86
  87       while not os.path.ismount(search):
  88         if s.src_likeparent(search):
  89           xinfo.reverse()
  90           if len(xinfo): infol.append('want=' + os.path.join(*xinfo))
  91           return search
  92
  93         ascend()
  94
  95     # no .git found anywhere
  96     return src
  97
  98   def src_prenormaliser(s, d, infol): # callers may monkey-patch away
  99     return os.path.join(s.cwd, os.path.abspath(d))
 100
 101   def srcdir_find_rune(s, d):
 102     script = s.find_rune_base
 103     for excl in s.excludes + [s.output_name, s.manifest_name]:
 104       assert("'" not in excl)
 105       script += r" \! -name '%s'" % excl
 106     script += ' -print0'
 107     return script
 108
 109   def manifest_append(s, name, infol):
 110     s._manifest.append((name, ' '.join(infol)))
 111
 112   def new_output_name(s, nametail, infol):
 113     s._outcounter += 1
 114     name = '%04d-%s' % (s._outcounter, nametail)
 115     s.manifest_append(name, infol)
 116     return name
 117
 118   def open_output_fh(s, name, mode):
 119     return open(os.path.join(s._destdir, name), mode)
 120
 121   def src_dir(s, d, infol):
 122     try: name = s._dirmap[d]
 123     except KeyError: pass
 124     else:
 125       s.manifest_append(name, infol)
 126       return
 127
 128     if s.show_pathnames: infol.append(d)
 129     find_rune = s.srcdir_find_rune(d)
 130     total_rune = s.rune_cpio % find_rune
 131
 132     name = s.new_output_name('src.cpio', infol)
 133     s._dirmap[d] = name
 134     fh = s.open_output_fh(name, 'wb')
 135
 136     subprocess.run(s.rune_shell + [total_rune],
 137                    cwd=d,
 138                    stdin=subprocess.DEVNULL,
 139                    stdout=fh,
 140                    restore_signals=True,
 141                    check=True)
 142     fh.close()
 143
 144   def src_indir(s, d, infol):
 145     d = s.src_prenormaliser(d, infol)
 146     if not s.src_filter(d): return
 147
 148     d = s.src_parentfinder(d, infol)
 149     s.dir(d, infol)
 150
 151   def report_from_packages_debian(s, files):
 152     dpkg_S_in = tempfile.TemporaryFile()
 153     for (file, infols) in files.items():
 154       assert('\n' not in file)
 155       dpkg_S_in.write(file)
 156       dpkg_S_in.write('\0')
 157     dpkg_S_in.seek(0)
 158     cmdl = ['xargs','-0r','dpkg','-S','--']
 159     dpkg_S = subprocess.Popen(cmdl,
 160                              cwd='/',
 161                              stdin=dpkg_S_in,
 162                              stdout=subprocess.PIPE,
 163                              close_fds=False)
 164     dpkg_show_in = tempfile.TemporaryFile()
 165     pkginfos = { }
 166     for l in dpkgs.stdout:
 167       (pkgs, fname) = l.split(': ',1)
 168       pkgs = pkgs.split(', ')
 169       for p in pkgs:
 170         pkginfos[
 171         print(p, file=dpkg_show_in)
 172
 173     dpkg-query --show PACKAGE
 174
 175   def thing_ought_packaged(s, fname):
 176     return s.thing_matches_globs(fname, s.src_package_globs)
 177
 178   def src_file_packaged(s, fname);
 179     try: s._package_files[fname].append(infol)
 180     except KeyError: s._package_files[fname] = [infol]
 181
 182   def src_file(s, fname, infol):
 183     def fngens():
 184       yield fname
 185       yield s.path_prenormaliser(fname)
 186       yield os.path.realpath(fname)
 187
 188     for fn in fngens():
 189       if s.thing_ought_packaged(fngen):
 190         s.src_file_packaged(fname, infol)
 191         return
 192
 193     s.src_indir(fname, infol)
 194
 195   def src_argv0(s, program, infol):
 196     s.src_file(s, program, infol)
 197
 198   def src_syspath(s, fname, infol):
 199     s.src_indir(fname, infol)
 200
 201   def src_module(s, m, infol):
 202     try: fname = m.__file__
 203     except AttributeError: return
 204     infol.append(m.__name__)
 205
 206     if s.thing_ought_packaged(fname):
 207       s.src_file_packaged(fname, infol)
 208     else:
 209       s.src_indir(s, fname)
 210
 211   def srcs_allitems(s, dirs=sys.path):
 212     s.src_argv0(sys.argv[0], ['argv[0]'])
 213     for d in sys.path:
 214       s.src_syspath(d, ['sys.path'])
 215     for m in sys.modules.values():
 216       s.src_module(m, ['sys.modules'])
 217     s.report_from_packages(s, s._package_files)
 218
 219   def mk_portmanteau(s):
 220     cmdl = s.rune_shell + [ s.rune_portmanteau, 'x',
 221                             s.output_name, s.manifest_name ]
 222     mfh = s.open_output_fh(s.manifest_name,'w')
 223     for (name, info) in s._manifest:
 224       if name is not None: cmdl.append(name)
 225       print('%s\t%s' % (name,info), file=mfh)
 226     mfh.close()
 227     subprocess.run(cmdl,
 228                    cwd=s._destdir,
 229                    stdin=subprocess.DEVNULL,
 230                    stdout=sys.stderr,
 231                    restore_signals=True,
 232                    check=True)
 233
 234   def generate(s):
 235     s.srcs_allitems()
 236     s.mk_portmanteau()