1 # Automatic source code provision (AGPL compliance)
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
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
29 # ^ by default, is find ... -print0
31 cpio -Hustar -o --quiet -0 -R 1000:1000 || \
32 cpio -Hustar -o --quiet -0
35 s.rune_portmanteau = r'''
38 GZIP=-1 tar zcf "$outfile" "$@"
40 s.manifest_name='0000-MANIFEST.txt'
46 s._package_files = { } # map filename => infol
48 def thing_matches_globs(s, thing, globs):
50 negate = pat.startswith('!')
51 if negate: pat = pat[1:]
52 if fnmatch.fnmatch(thing, pat):
56 def src_filter_glob(s, src): # default s.src_filter
57 return s.thing_matches_globs(s, src, s.src_filter_globs)
59 def src_likeparent_git(s, src):
61 os.stat(os.path.join(src, '.git/.'))
62 except FileNotFoundError:
67 def src_parentfinder(s, src, infol): # callers may monkey-patch away
68 for deref in (False,True):
73 search = os.path.realpath(search)
77 xinfo.append(os.path.basename(search))
78 search = os.path.dirname(search)
81 stab = os.lstat(search)
82 except FileNotFoundError:
84 if stat.S_ISREG(stab.st_mode):
87 while not os.path.ismount(search):
88 if s.src_likeparent(search):
90 if len(xinfo): infol.append('want=' + os.path.join(*xinfo))
95 # no .git found anywhere
98 def src_prenormaliser(s, d, infol): # callers may monkey-patch away
99 return os.path.join(s.cwd, os.path.abspath(d))
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
109 def manifest_append(s, name, infol):
110 s._manifest.append({ 'file':name, 'info':' '.join(infol) })
112 def new_output_name(s, nametail, infol):
114 name = '%04d-%s' % (s._outcounter, nametail)
115 s.manifest_append(name, infol)
118 def open_output_fh(s, name, mode):
119 return open(os.path.join(s._destdir, name), mode)
121 def src_dir(s, d, infol):
122 try: name = s._dirmap[d]
123 except KeyError: pass
125 s.manifest_append(name, infol)
128 if s.show_pathnames: infol.append(d)
129 find_rune = s.srcdir_find_rune(d)
130 total_rune = s.rune_cpio % find_rune
132 name = s.new_output_name('src.cpio', infol)
134 fh = s.open_output_fh(name, 'wb')
136 subprocess.run(s.rune_shell + [total_rune],
138 stdin=subprocess.DEVNULL,
140 restore_signals=True,
144 def src_indir(s, d, infol):
145 d = s.src_prenormaliser(d, infol)
146 if not s.src_filter(d): return
148 d = s.src_parentfinder(d, infol)
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')
158 cmdl = ['xargs','-0r','dpkg','-S','--']
159 dpkg_S = subprocess.Popen(cmdl,
162 stdout=subprocess.PIPE,
165 dpkg_show_in = tempfile.TemporaryFile()
167 for l in dpkgs.stdout:
168 (pkgs, fname) = l.split(': ',1)
169 pkgs = pkgs.split(', ')
171 pkginfos.setdefault(p,{'files':[]})['files'].append([fname, infol])
172 print(p, file=dpkg_show_in)
173 assert(dpkg_S.wait() == 0)
175 cmdl = ['xargs','-r','dpkg-query',
176 r'-f${Package}\t${Architecture}\t${Version}\t${Source}\t${Source-Version}\n'
178 dpkg_show = subprocess.Popen(cmdl,
181 stdout=subprocess.PIPE,
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)
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) })
201 def thing_ought_packaged(s, fname):
202 return s.thing_matches_globs(fname, s.src_package_globs)
204 def src_file_packaged(s, fname);
205 s._package_files.setdefault(fname,[]).append(infol)
207 def src_file(s, fname, infol):
210 yield s.path_prenormaliser(fname)
211 yield os.path.realpath(fname)
214 if s.thing_ought_packaged(fngen):
215 s.src_file_packaged(fname, infol)
218 s.src_indir(fname, infol)
220 def src_argv0(s, program, infol):
221 s.src_file(s, program, infol)
223 def src_syspath(s, fname, infol):
224 s.src_indir(fname, infol)
226 def src_module(s, m, infol):
227 try: fname = m.__file__
228 except AttributeError: return
229 infol.append(m.__name__)
231 if s.thing_ought_packaged(fname):
232 s.src_file_packaged(fname, infol)
234 s.src_indir(s, fname)
236 def srcs_allitems(s, dirs=sys.path):
237 s.src_argv0(sys.argv[0], ['argv[0]'])
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)
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)
256 stdin=subprocess.DEVNULL,
258 restore_signals=True,