Add a script for modifying ISO images

With this script it's possible to add additional files into an ISO file.
If the file happens to be ks.cfg, the boot configs are tweaked so that
the kickstart is actually used.

Resolves: #503
Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>

bin/pungi-patch-iso

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <https://gnu.org/licenses/>.
+
+import argparse
+import logging
+import os
+import sys
+
+here = sys.path[0]
+if here != '/usr/bin':
+    # Git checkout
+    sys.path[0] = os.path.dirname(here)
+
+from pungi_utils import patch_iso
+
+
+def main(args=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-v', '--verbose', action='store_true',
+                        help='Print debugging information')
+    parser.add_argument('--supported', choices=('true', 'false'),
+                        help='Override supported bit on the ISO')
+    parser.add_argument('--volume-id',
+                        help='Override volume ID on the ISO')
+    parser.add_argument('--force-arch',
+                        help='Treat the ISO as bootable on given architecture')
+    parser.add_argument('target', metavar='TARGET_ISO',
+                        help='which file to write the result to')
+    parser.add_argument('source', metavar='SOURCE_ISO',
+                        help='source ISO to work with')
+    parser.add_argument('dir', metavar='GRAFT_DIR',
+                        help='extra directory to graft on the ISO')
+    opts = parser.parse_args(args)
+
+    level = logging.DEBUG if opts.verbose else logging.INFO
+    format = '%(levelname)s: %(message)s'
+    logging.basicConfig(level=level, format=format)
+    log = logging.getLogger()
+
+    patch_iso.run(log, opts)
+
+
+if __name__ == '__main__':
+    if main():
+        sys.exit(1)

pungi.spec

@@ -84,6 +84,7 @@ rm -rf %{buildroot}
 %{_bindir}/%{name}-create-unified-isos
 %{_bindir}/%{name}-config-validate
 %{_bindir}/%{name}-fedmsg-notification
+%{_bindir}/%{name}-patch-iso
 
 %check
 nosetests --exe

pungi_utils/patch_iso.py

@@ -0,0 +1,133 @@
+# -*- coding: utf-8 -*-
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <https://gnu.org/licenses/>.
+
+from kobo import shortcuts
+import os
+import productmd
+import tempfile
+try:
+    from shlex import quote
+except ImportError:
+    from pipes import quote
+
+from pungi import util
+from pungi.phases.buildinstall import tweak_configs
+from pungi.wrappers import iso
+
+
+def sh(log, cmd, *args, **kwargs):
+    log.info('Running: %s', ' '.join(quote(x) for x in cmd))
+    ret, out = shortcuts.run(cmd, *args, **kwargs)
+    if out:
+        log.debug('%s', out)
+    return ret, out
+
+
+def get_lorax_dir(default='/usr/share/lorax'):
+    try:
+        _, out = shortcuts.run(['python3', '-c' 'import pylorax; print(pylorax.find_templates())'])
+        return out.strip()
+    except Exception:
+        return default
+
+
+def as_bool(arg):
+    if arg == 'true':
+        return True
+    elif arg == 'false':
+        return False
+    else:
+        return arg
+
+
+def get_arch(log, iso_dir):
+    di_path = os.path.join(iso_dir, '.discinfo')
+    if os.path.exists(di_path):
+        di = productmd.discinfo.DiscInfo()
+        di.load(di_path)
+        log.info('Detected bootable ISO for %s (based on .discinfo)', di.arch)
+        return di.arch
+
+    ti_path = os.path.join(iso_dir, '.treeinfo')
+    if os.path.exists(ti_path):
+        ti = productmd.treeinfo.TreeInfo()
+        ti.load(ti_path)
+        log.info('Detected bootable ISO for %s (based on .treeinfo)', ti.tree.arch)
+        return ti.tree.arch
+
+    # There is no way to tell the architecture of an ISO file without guessing.
+    # Let's print a warning and continue with assuming unbootable ISO.
+
+    log.warning('Failed to detect arch for ISO, assuming unbootable one.')
+    log.warning('If this is incorrect, use the --force-arch option.')
+    return None
+
+
+def run(log, opts):
+    # mount source iso
+    log.info('Mounting %s', opts.source)
+    target = os.path.abspath(opts.target)
+
+    with util.temp_dir(prefix='patch-iso-') as work_dir:
+        with iso.mount(opts.source) as source_iso_dir:
+            util.copy_all(source_iso_dir, work_dir)
+
+        # Make everything writable
+        for root, dirs, files in os.walk(work_dir):
+            for name in files:
+                os.chmod(os.path.join(root, name), 0o640)
+            for name in dirs:
+                os.chmod(os.path.join(root, name), 0o755)
+
+        # volume id is copied from source iso unless --label is specified
+        volume_id = opts.volume_id or iso.get_volume_id(opts.source)
+
+        # create graft points from mounted source iso + overlay dir
+        graft_points = iso.get_graft_points([work_dir, opts.dir])
+        # if ks.cfg is detected, patch syslinux + grub to use it
+        if 'ks.cfg' in graft_points:
+            log.info('Adding ks.cfg to boot configs')
+            tweak_configs(work_dir, volume_id, graft_points['ks.cfg'])
+
+        arch = opts.force_arch or get_arch(log, work_dir)
+
+        with tempfile.NamedTemporaryFile(prefix='graft-points-') as graft_file:
+            iso.write_graft_points(graft_file.name, graft_points,
+                                   exclude=["*/TRANS.TBL", "*/boot.cat"])
+
+            # make the target iso bootable if source iso is bootable
+            boot_args = input_charset = None
+            if arch:
+                boot_args = iso.get_boot_options(
+                    arch, os.path.join(get_lorax_dir(), 'config_files/ppc'))
+                input_charset = 'utf-8' if 'ppc' not in arch else None
+            # Create the target ISO
+            mkisofs_cmd = iso.get_mkisofs_cmd(target, None,
+                                              volid=volume_id,
+                                              exclude=["./lost+found"],
+                                              graft_points=graft_file.name,
+                                              input_charset=input_charset,
+                                              boot_args=boot_args)
+            sh(log, mkisofs_cmd, workdir=work_dir)
+
+    # isohybrid support
+    if arch in ["x86_64", "i386"]:
+        isohybrid_cmd = iso.get_isohybrid_cmd(target, arch)
+        sh(log, isohybrid_cmd)
+
+    supported = as_bool(opts.supported or iso.get_checkisomd5_data(opts.source)['Supported ISO'])
+    # implantmd5 + supported bit (use the same as on source iso, unless
+    # overriden by --supported option)
+    isomd5sum_cmd = iso.get_implantisomd5_cmd(target, supported)
+    sh(log, isomd5sum_cmd)

setup.py

@@ -41,6 +41,7 @@ setup(
         'bin/pungi-fedmsg-notification',
         'bin/pungi-koji',
         'bin/pungi-make-ostree',
+        'bin/pungi-patch-iso',
     ],
     data_files      = [
         ('/usr/share/pungi', glob.glob('share/*.xsl')),

tests/test_patch_iso.py

@@ -0,0 +1,228 @@
+# -*- coding: utf-8 -*-
+
+import mock
+import os
+import sys
+try:
+    import unittest2 as unittest
+except ImportError:
+    import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from tests.helpers import boom, touch, copy_fixture
+from pungi_utils import patch_iso
+
+
+class TestUnifiedIsos(unittest.TestCase):
+    pass
+
+
+class TestGetLoraxDir(unittest.TestCase):
+    @mock.patch('kobo.shortcuts.run')
+    def test_success(self, mock_run):
+        mock_run.return_value = (0, 'hello')
+        self.assertEqual(patch_iso.get_lorax_dir(None), 'hello')
+        self.assertEqual(1, len(mock_run.call_args_list))
+
+    @mock.patch('kobo.shortcuts.run')
+    def test_crash(self, mock_run):
+        mock_run.side_effect = boom
+        self.assertEqual(patch_iso.get_lorax_dir('hello'), 'hello')
+        self.assertEqual(1, len(mock_run.call_args_list))
+
+
+class TestSh(unittest.TestCase):
+    @mock.patch('kobo.shortcuts.run')
+    def test_cmd(self, mock_run):
+        mock_run.return_value = (0, 'ok')
+        log = mock.Mock()
+        patch_iso.sh(log, ['ls'], foo='bar')
+        self.assertEqual(mock_run.call_args_list,
+                         [mock.call(['ls'], foo='bar')])
+        self.assertEqual(log.info.call_args_list,
+                         [mock.call('Running: %s', 'ls')])
+        self.assertEqual(log.debug.call_args_list,
+                         [mock.call('%s', 'ok')])
+
+
+class TestAsBool(unittest.TestCase):
+    def test_true(self):
+        self.assertTrue(patch_iso.as_bool('true'))
+
+    def test_false(self):
+        self.assertFalse(patch_iso.as_bool('false'))
+
+    def test_anything_else(self):
+        obj = mock.Mock()
+        self.assertIs(patch_iso.as_bool(obj), obj)
+
+
+class EqualsAny(object):
+    def __eq__(self, another):
+        return True
+
+    def __repr__(self):
+        return u'ANYTHING'
+
+ANYTHING = EqualsAny()
+
+
+class TestPatchingIso(unittest.TestCase):
+
+    @mock.patch('pungi_utils.patch_iso.util.copy_all')
+    @mock.patch('pungi_utils.patch_iso.iso')
+    @mock.patch('pungi_utils.patch_iso.sh')
+    def test_whole(self, sh, iso, copy_all):
+        iso.mount.return_value.__enter__.return_value = 'mounted-iso-dir'
+
+        def _create_files(src, dest):
+            touch(os.path.join(dest, 'dir', 'file.txt'), 'Hello')
+
+        copy_all.side_effect = _create_files
+
+        log = mock.Mock(name='logger')
+        opts = mock.Mock(
+            target='test.iso',
+            source='source.iso',
+            force_arch=None,
+            volume_id='FOOBAR',
+        )
+        patch_iso.run(log, opts)
+
+        self.assertEqual(iso.get_mkisofs_cmd.call_args_list,
+                         [mock.call(os.path.abspath(opts.target), None,
+                                    boot_args=None,
+                                    exclude=['./lost+found'],
+                                    graft_points=ANYTHING,
+                                    input_charset=None,
+                                    volid='FOOBAR')])
+        self.assertEqual(iso.mount.call_args_list,
+                         [mock.call('source.iso')])
+        self.assertEqual(copy_all.mock_calls,
+                         [mock.call('mounted-iso-dir', ANYTHING)])
+        self.assertEqual(
+            sh.call_args_list,
+            [mock.call(log, iso.get_mkisofs_cmd.return_value, workdir=ANYTHING),
+             mock.call(log, iso.get_implantisomd5_cmd.return_value)])
+
+    @mock.patch('pungi_utils.patch_iso.util.copy_all')
+    @mock.patch('pungi_utils.patch_iso.iso')
+    @mock.patch('pungi_utils.patch_iso.sh')
+    def test_detect_arch_discinfo(self, sh, iso, copy_all):
+        iso.mount.return_value.__enter__.return_value = 'mounted-iso-dir'
+
+        def _create_files(src, dest):
+            touch(os.path.join(dest, 'dir', 'file.txt'), 'Hello')
+            touch(os.path.join(dest, '.discinfo'),
+                  '1487578537.111417\nDummy Product 1.0\nppc64\n1')
+
+        copy_all.side_effect = _create_files
+
+        log = mock.Mock(name='logger')
+        opts = mock.Mock(
+            target='test.iso',
+            source='source.iso',
+            force_arch=None,
+            volume_id=None
+        )
+        patch_iso.run(log, opts)
+
+        self.assertEqual(iso.mount.call_args_list,
+                         [mock.call('source.iso')])
+        self.assertEqual(iso.get_mkisofs_cmd.call_args_list,
+                         [mock.call(os.path.abspath(opts.target), None,
+                                    boot_args=iso.get_boot_options.return_value,
+                                    exclude=['./lost+found'],
+                                    graft_points=ANYTHING,
+                                    input_charset=None,
+                                    volid=iso.get_volume_id.return_value)])
+        self.assertEqual(copy_all.mock_calls,
+                         [mock.call('mounted-iso-dir', ANYTHING)])
+        self.assertEqual(
+            sh.call_args_list,
+            [mock.call(log, iso.get_mkisofs_cmd.return_value, workdir=ANYTHING),
+             mock.call(log, iso.get_implantisomd5_cmd.return_value)])
+
+    @mock.patch('pungi_utils.patch_iso.util.copy_all')
+    @mock.patch('pungi_utils.patch_iso.iso')
+    @mock.patch('pungi_utils.patch_iso.sh')
+    def test_run_isohybrid(self, sh, iso, copy_all):
+        iso.mount.return_value.__enter__.return_value = 'mounted-iso-dir'
+
+        def _create_files(src, dest):
+            touch(os.path.join(dest, 'dir', 'file.txt'), 'Hello')
+            copy_fixture(
+                'DP-1.0-20161013.t.4/compose/Server/x86_64/os/.treeinfo',
+                os.path.join(dest, '.treeinfo')
+            )
+
+        copy_all.side_effect = _create_files
+
+        log = mock.Mock(name='logger')
+        opts = mock.Mock(
+            target='test.iso',
+            source='source.iso',
+            force_arch=None,
+            volume_id=None
+        )
+        patch_iso.run(log, opts)
+
+        self.assertEqual(iso.mount.call_args_list,
+                         [mock.call('source.iso')])
+        self.assertEqual(iso.get_mkisofs_cmd.call_args_list,
+                         [mock.call(os.path.abspath(opts.target), None,
+                                    boot_args=iso.get_boot_options.return_value,
+                                    exclude=['./lost+found'],
+                                    graft_points=ANYTHING,
+                                    input_charset='utf-8',
+                                    volid=iso.get_volume_id.return_value)])
+        self.assertEqual(copy_all.mock_calls,
+                         [mock.call('mounted-iso-dir', ANYTHING)])
+        self.assertEqual(
+            sh.call_args_list,
+            [mock.call(log, iso.get_mkisofs_cmd.return_value, workdir=ANYTHING),
+             mock.call(log, iso.get_isohybrid_cmd.return_value),
+             mock.call(log, iso.get_implantisomd5_cmd.return_value)])
+
+    @mock.patch('pungi_utils.patch_iso.tweak_configs')
+    @mock.patch('pungi_utils.patch_iso.util.copy_all')
+    @mock.patch('pungi_utils.patch_iso.iso')
+    @mock.patch('pungi_utils.patch_iso.sh')
+    def test_add_ks_cfg(self, sh, iso, copy_all, tweak_configs):
+        iso.mount.return_value.__enter__.return_value = 'mounted-iso-dir'
+        iso.get_graft_points.return_value = {
+            'ks.cfg': 'path/to/ks.cfg',
+        }
+
+        def _create_files(src, dest):
+            touch(os.path.join(dest, 'dir', 'file.txt'), 'Hello')
+
+        copy_all.side_effect = _create_files
+
+        log = mock.Mock(name='logger')
+        opts = mock.Mock(
+            target='test.iso',
+            source='source.iso',
+            force_arch='s390',
+            volume_id='foobar',
+        )
+        patch_iso.run(log, opts)
+
+        self.assertEqual(iso.mount.call_args_list,
+                         [mock.call('source.iso')])
+        self.assertEqual(iso.get_mkisofs_cmd.call_args_list,
+                         [mock.call(os.path.abspath(opts.target), None,
+                                    boot_args=iso.get_boot_options.return_value,
+                                    exclude=['./lost+found'],
+                                    graft_points=ANYTHING,
+                                    input_charset='utf-8',
+                                    volid='foobar')])
+        self.assertEqual(tweak_configs.call_args_list,
+                         [mock.call(ANYTHING, 'foobar', 'path/to/ks.cfg')])
+        self.assertEqual(copy_all.mock_calls,
+                         [mock.call('mounted-iso-dir', ANYTHING)])
+        self.assertEqual(
+            sh.call_args_list,
+            [mock.call(log, iso.get_mkisofs_cmd.return_value, workdir=ANYTHING),
+             mock.call(log, iso.get_implantisomd5_cmd.return_value)])