+ #!/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)

+ %{_bindir}/%{name}-patch-iso

+ from pungi.util import get_file_size, get_mtime, failable, makedirs

+ BOOT_CONFIGS = [

+     "isolinux/isolinux.cfg",

+     "etc/yaboot.conf",

+     "ppc/ppc64/yaboot.conf",

+     "EFI/BOOT/BOOTX64.conf",

+     "EFI/BOOT/grub.cfg",

+ ]

+ 

+ 

+ def tweak_configs(path, volid, ks_file, configs=BOOT_CONFIGS):

+     volid_escaped = volid.replace(" ", r"\x20").replace("\\", "\\\\")

+     volid_escaped_2 = volid_escaped.replace("\\", "\\\\")

+     found_configs = []

+     for config in configs:

+         config_path = os.path.join(path, config)

+         if not os.path.exists(config_path):

+             continue

+         found_configs.append(config)

+ 

+         with open(config_path, "r") as f:

+             data = f.read()

+         os.unlink(config_path)  # break hadlink by removing file writing a new one

+ 

+         # double-escape volid in yaboot.conf

+         new_volid = volid_escaped_2 if 'yaboot' in config else volid_escaped

+ 

+         ks = (" ks=hd:LABEL=%s:/ks.cfg" % new_volid) if ks_file else ""

+ 

+         # pre-f18

+         data = re.sub(r":CDLABEL=[^ \n]*", r":CDLABEL=%s%s" % (new_volid, ks), data)

+         # f18+

+         data = re.sub(r":LABEL=[^ \n]*", r":LABEL=%s%s" % (new_volid, ks), data)

+         data = re.sub(r"(search .* -l) '[^'\n]*'", r"\1 '%s'" % volid, data)

+ 

+         with open(config_path, "w") as f:

+             f.write(data)

+ 

+     return found_configs

+ 

+ 

+     found_configs = tweak_configs(tmp_dir, volid, kickstart_file)

+     if kickstart_file and found_configs:

+         shutil.copy2(kickstart_file, os.path.join(dst, "ks.cfg"))

+ 

+         with iso.mount(image, logger=compose._logger) as mount_tmp_dir:

+             for config in BOOT_CONFIGS:

+                 config_path = os.path.join(tmp_dir, config)

+                 config_in_image = os.path.join(mount_tmp_dir, config)

+ 

+                 if os.path.isfile(config_in_image):

+                     cmd = ["cp", "-v", "--remove-destination", config_path, config_in_image]

+                     run(cmd)

+ import contextlib

+ from pungi import util

+ def get_checkisomd5_data(iso_path, logger=None):

+     items = [line.strip().rsplit(":", 1) for line in output.splitlines()]

+     items = dict([(k, v.strip()) for k, v in items])

+     md5 = items.get(iso_path, '')

+     return items

+ 

+ 

+ def get_implanted_md5(iso_path, logger=None):

+     return (get_checkisomd5_data(iso_path, logger=logger) or {}).get(iso_path)

+ 

+ 

+ @contextlib.contextmanager

+ def mount(image, logger=None):

+     """Mount an image and make sure it's unmounted.

+ 

+     The yielded path will only be valid in the with block and is removed once

+     the image is unmounted.

+     """

+     # use guestmount to mount the image, which doesn't require root privileges

+     # LIBGUESTFS_BACKEND=direct: running qemu directly without libvirt

+     with util.temp_dir(prefix='iso-mount-') as mount_dir:

+         run(["LIBGUESTFS_BACKEND=direct", "guestmount", "-a", image, "-m", "/dev/sda", mount_dir])

+         try:

+             yield mount_dir

+         finally:

+             util.fusermount(mount_dir, logger=logger)

+ # -*- 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)

+         'bin/pungi-patch-iso',

+ from pungi.phases.buildinstall import (BuildinstallPhase, BuildinstallThread, link_boot_iso,

+                                        BOOT_CONFIGS, tweak_configs)

+ class TestTweakConfigs(PungiTestCase):

+ 

+     def test_tweak_configs(self):

+         configs = []

+         for cfg in BOOT_CONFIGS:

+             if 'yaboot' not in cfg:

+                 configs.append(os.path.join(self.topdir, cfg))

+                 touch(configs[-1], ':LABEL=baz')

+         tweak_configs(self.topdir, 'new volid', os.path.join(self.topdir, 'ks.cfg'))

+         for cfg in configs:

+             with open(cfg) as f:

+                 self.assertEqual(

+                     f.read().strip(),

+                     ':LABEL=new\\x20volid ks=hd:LABEL=new\\x20volid:/ks.cfg')

+ 

+     def test_tweak_configs_yaboot(self):

+         configs = []

+         for cfg in BOOT_CONFIGS:

+             if 'yaboot' in cfg:

+                 configs.append(os.path.join(self.topdir, cfg))

+                 touch(configs[-1], ':LABEL=baz')

+         tweak_configs(self.topdir, 'new volid', os.path.join(self.topdir, 'ks.cfg'))

+         for cfg in configs:

+             with open(os.path.join(self.topdir, cfg)) as f:

+                 self.assertEqual(

+                     f.read().strip(),

+                     ':LABEL=new\\\\x20volid ks=hd:LABEL=new\\\\x20volid:/ks.cfg')

+ 

+ 

+     @mock.patch('pungi.util.run_unmount_cmd')

+     @mock.patch('pungi.wrappers.iso.run')

+     def test_mount_iso(self, mock_run, mock_unmount):

+         with iso.mount('dummy') as temp_dir:

+             self.assertTrue(os.path.isdir(temp_dir))

+         self.assertEqual(len(mock_run.call_args_list), 1)

+         self.assertEqual(len(mock_unmount.call_args_list), 1)

+         self.assertFalse(os.path.isdir(temp_dir))

+ 

+     @mock.patch('pungi.util.run_unmount_cmd')

+     @mock.patch('pungi.wrappers.iso.run')

+     def test_mount_iso_always_unmounts(self, mock_run, mock_unmount):

+         try:

+             with iso.mount('dummy') as temp_dir:

+                 self.assertTrue(os.path.isdir(temp_dir))

+                 raise RuntimeError()

+         except RuntimeError:

+             pass

+         self.assertEqual(len(mock_run.call_args_list), 1)

+         self.assertEqual(len(mock_unmount.call_args_list), 1)

+         self.assertFalse(os.path.isdir(temp_dir))

+ # -*- 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)])