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({ 'file':name, 'info':' '.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                               stderr=sys.stderr,
 164                               close_fds=False)
 165     dpkg_show_in = tempfile.TemporaryFile()
 166     pkginfos = { }
 167     for l in dpkgs.stdout:
 168       (pkgs, fname) = l.split(': ',1)
 169       pkgs = pkgs.split(', ')
 170       for p in pkgs:
 171         pkginfos.setdefault(p,{'files':[]})['files'].append([fname, infol])
 172         print(p, file=dpkg_show_in)
 173     assert(dpkg_S.wait() == 0)
 174     dpkg_show_in.seek(0)
 175     cmdl = ['xargs','-r','dpkg-query',
 176  r'-f${Package}\t${Architecture}\t${Version}\t${Source}\t${Source-Version}\n'
 177             '--show','--']
 178     dpkg_show = subprocess.Popen(cmdl,
 179                                  cwd='/',
 180                                  stdin=dpkg_show_in,
 181                                  stdout=subprocess.PIPE,
 182                                  stderr=sys.stderr,
 183                                  close_fds=False)
 184     for l in dpkg_show.stdout:
 185       (p,a,v) = l.split('\t')
 186       pkginfos[p]['arch'] = a
 187       pkginfos[p]['version'] = v
 188       pkginfos[p]['source'] = v
 189       pkginfos[p]['sourceversion'] = v
 190     assert(dpkg_show.wait() == 0)
 191     pl = pkginfos.keys()
 192     pl.sort()
 193     for p in pl:
 194       pi = pkginfos[p]
 195       debfname = '%s_%s_%s.deb' % (p, pi['version'], pi['arch'])
 196       dscfname = '%s_%s.dsc' % (pi['source'], pi['sourceversion'])
 197       s._manifest.append({ 'file_print': dscfname, 'info': debfname })
 198       for (fname, infol) in pi['files']:
 199         s._manifest.append({ 'file_print': fname, 'info': ' '.join(infol) })
 200
 201   def thing_ought_packaged(s, fname):
 202     return s.thing_matches_globs(fname, s.src_package_globs)
 203
 204   def src_file_packaged(s, fname);
 205     s._package_files.setdefault(fname,[]).append(infol)
 206
 207   def src_file(s, fname, infol):
 208     def fngens():
 209       yield fname
 210       yield s.path_prenormaliser(fname)
 211       yield os.path.realpath(fname)
 212
 213     for fn in fngens():
 214       if s.thing_ought_packaged(fngen):
 215         s.src_file_packaged(fname, infol)
 216         return
 217
 218     s.src_indir(fname, infol)
 219
 220   def src_argv0(s, program, infol):
 221     s.src_file(s, program, infol)
 222
 223   def src_syspath(s, fname, infol):
 224     s.src_indir(fname, infol)
 225
 226   def src_module(s, m, infol):
 227     try: fname = m.__file__
 228     except AttributeError: return
 229     infol.append(m.__name__)
 230
 231     if s.thing_ought_packaged(fname):
 232       s.src_file_packaged(fname, infol)
 233     else:
 234       s.src_indir(s, fname)
 235
 236   def srcs_allitems(s, dirs=sys.path):
 237     s.src_argv0(sys.argv[0], ['argv[0]'])
 238     for d in sys.path:
 239       s.src_syspath(d, ['sys.path'])
 240     for m in sys.modules.values():
 241       s.src_module(m, ['sys.modules'])
 242     s.report_from_packages(s, s._package_files)
 243
 244   def mk_portmanteau(s):
 245     cmdl = s.rune_shell + [ s.rune_portmanteau, 'x',
 246                             s.output_name, s.manifest_name ]
 247     mfh = s.open_output_fh(s.manifest_name,'w')
 248     for me in s._manifest:
 249       try: fname = me['file']
 250       except KeyError: fname = me.get('file_print','')
 251       else: cmdl.append(name)
 252       print('%s\t%s' % (fname, me['info']), file=mfh)
 253     mfh.close()
 254     subprocess.run(cmdl,
 255                    cwd=s._destdir,
 256                    stdin=subprocess.DEVNULL,
 257                    stdout=sys.stderr,
 258                    restore_signals=True,
 259                    check=True)
 260
 261   def generate(s):
 262     s.srcs_allitems()
 263     s.mk_portmanteau()