Skip to content
Snippets Groups Projects
populate-depot.py 49.8 KiB
Newer Older
            os.path.join(self.depot, runtime.build_id_file), 'w',
        ) as writer:
            writer.write(f'{runtime.version}\n')

        if runtime.include_sdk:
            with open(
                os.path.join(self.depot, runtime.sdk_build_id_file), 'w',
            ) as writer:
                writer.write(f'{runtime.version}\n')

        if self.unpack_sources:
            with open(
                os.path.join(runtime.path, runtime.sources), 'rb',
            ) as reader:
                for stanza in Sources.iter_paragraphs(
                    sequence=reader,
                    use_apt_pkg=True,
                ):
                    if stanza['package'] in self.unpack_sources:
                        for f in stanza['files']:
                            name = f['name']

                            if name.endswith('.dsc'):
                                dest = os.path.join(
                                    self.unpack_sources_into,
                                    runtime.name,
                                    stanza['package'],
                                )

                                with suppress(FileNotFoundError):
                                    logger.info('Removing %r', dest)
                                    shutil.rmtree(dest)

                                        os.path.join(
                                            runtime.path,
                                            'sources',
                                            f['name'],
                                        ),
                                    ],
                                    check=True,
                                )

    def download_runtime(self, runtime: Runtime) -> None:
        """
        Download a pre-prepared Platform from a previous container
        runtime build.
        """

        pinned = runtime.pin_version(self.opener)
        for basename in runtime.runtime_files:
            downloaded = runtime.fetch(basename, self.opener)
            if self.include_archives:
                os.link(downloaded, os.path.join(self.depot, basename))
                os.path.join(self.depot, runtime.build_id_file), 'w',
            ) as writer:
                writer.write(f'{pinned}\n')

            if runtime.include_sdk:
                with open(
                    os.path.join(self.depot, runtime.sdk_build_id_file), 'w',
                ) as writer:
                    writer.write(f'{pinned}\n')

        if self.unpack_sources:
            with tempfile.TemporaryDirectory(prefix='populate-depot.') as tmp:
                want = set(self.unpack_sources)
                downloaded = runtime.fetch(
                with open(downloaded, 'rb') as reader:
                    for stanza in Sources.iter_paragraphs(
                        sequence=reader,
                        use_apt_pkg=True,
                    ):
                        if stanza['package'] in self.unpack_sources:
                            logger.info(
                                'Found %s in %s',
                                stanza['package'], runtime.name,
                            )
                            want.discard(stanza['package'])
                            os.makedirs(
                                os.path.join(self.cache or tmp, 'sources'),
                                exist_ok=True,
                            )

                            for f in stanza['files']:
                                name = f['name']
                                downloaded = runtime.fetch(
                                    os.path.join('sources', name),
                                    self.opener,
                                )

                            for f in stanza['files']:
                                name = f['name']

                                if name.endswith('.dsc'):
                                    dest = os.path.join(
                                        self.unpack_sources_into,
                                        runtime.name,
                                        stanza['package'],
                                    )

                                    with suppress(FileNotFoundError):
                                        logger.info('Removing %r', dest)
                                        shutil.rmtree(dest)

                                        ],
                                        check=True,
                                    )

                if want:
                    logger.warning(
                        'Did not find source package(s) %s in %s',
                        ', '.join(want), runtime.name,
                    )

    def download_scout_tarball(self, runtime: Runtime) -> None:
        """
        Download a pre-prepared LD_LIBRARY_PATH Steam Runtime from a
        previous scout build.
        """
        filename = 'steam-runtime.tar.xz'

        pinned = runtime.pin_version(self.opener)
        logger.info('Downloading steam-runtime build %s', pinned)
        os.makedirs(self.unpack_ld_library_path, exist_ok=True)

        downloaded = runtime.fetch(
            filename,
            self.opener,
        )
        subprocess.run(
            [
                'tar', '-C', self.unpack_ld_library_path, '-xf',
                downloaded,
            ],
            check=True,
        )
    def octal_escape_char(self, match: 're.Match') -> str:
        ret = []    # type: List[str]

        for byte in match.group(0).encode('utf-8', 'surrogateescape'):
            ret.append('\\%03o' % byte)

        return ''.join(ret)

    _NEEDS_OCTAL_ESCAPE = re.compile(r'[^-A-Za-z0-9+,./:@_]')

    def octal_escape(self, s: str) -> str:
        return self._NEEDS_OCTAL_ESCAPE.sub(self.octal_escape_char, s)

    def filename_is_windows_friendly(self, s: str) -> bool:
        for c in s:
            # This is the set of characters that are reserved in Windows
            # filenames, excluding '/' which obviously we're fine with
            # using as a directory separator.
            if c in r'<>:"\|?*':
                return False

            if c >= '\uDC80' and c <= '\uDCFF':
                # surrogate escape, not Unicode
                return False

        return True

    def write_lookaside(self, archive: str, dest: str) -> None:
        with tarfile.open(
            archive, mode='r'
        ) as unarchiver, gzip.open(
            os.path.join(dest, 'usr-mtree.txt.gz'), 'wt'
        ) as writer:
            lc_names = {}                   # type: Dict[str, str]
            differ_only_by_case = set()     # type: Set[str]
            not_windows_friendly = set()    # type: Set[str]
            sha256 = {}                     # type: Dict[str, str]
            sizes = {}                      # type: Dict[str, int]

            writer.write('#mtree\n')
            writer.write('. type=dir\n')

            for member in unarchiver:
                name = member.name

                if name.startswith('./'):
                    name = name[len('./'):]

                if not name.startswith('files/'):
                    continue

                name = name[len('files/'):]

                if not self.filename_is_windows_friendly(name):
                    not_windows_friendly.add(name)

                if name.lower() in lc_names:
                    differ_only_by_case.add(lc_names[name.lower()])
                    differ_only_by_case.add(name)
                else:
                    lc_names[name.lower()] = name

                fields = ['./' + self.octal_escape(name)]

                if member.isfile() or member.islnk():
                    fields.append('type=file')
                    fields.append('mode=%o' % member.mode)
                    fields.append('time=%d' % member.mtime)

                    if member.islnk():
                        writer.write(
                            '# hard link to {}\n'.format(
                                self.octal_escape(member.linkname),
                            ),
                        )
                        assert member.linkname in sizes
                        fields.append('size=%d' % sizes[member.linkname])

                        if sizes[member.linkname] > 0:
                            assert member.linkname in sha256
                            fields.append('sha256=' + sha256[member.linkname])
                    else:
                        fields.append('size=%d' % member.size)
                        sizes[member.name] = member.size

                        if member.size > 0:
                            hasher = hashlib.sha256()
                            extract = unarchiver.extractfile(member)
                            assert extract is not None
                            with extract:
                                while True:
                                    blob = extract.read(4096)

                                    if not blob:
                                        break

                                    hasher.update(blob)

                            fields.append('sha256=' + hasher.hexdigest())
                            sha256[member.name] = hasher.hexdigest()

                elif member.issym():
                    fields.append('type=link')
                    fields.append('link=' + self.octal_escape(member.linkname))
                elif member.isdir():
                    fields.append('type=dir')
                else:
                    writer.write(
                        '# unknown file type: {}\n'.format(
                            self.octal_escape(member.name),
                        ),
                    )
                    continue

                writer.write(' '.join(fields) + '\n')

            if differ_only_by_case:
                writer.write('\n')
                writer.write('# Files whose names differ only by case:\n')

                for name in sorted(differ_only_by_case):
                    writer.write('# {}\n'.format(self.octal_escape(name)))

            if not_windows_friendly:
                writer.write('\n')
                writer.write('# Files whose names are not Windows-friendly:\n')

                for name in sorted(not_windows_friendly):
                    writer.write('# {}\n'.format(self.octal_escape(name)))


def main() -> None:
    logging.basicConfig()
    logging.getLogger().setLevel(logging.DEBUG)

    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument(
        '--architecture', default='amd64,i386',
        help=(
            'Default dpkg architecture or comma-separated list of '
            'architectures'
        )
    )
    parser.add_argument(
        '--suite', default='',
        help=(
            'Default suite to use if none is specified'
        )
    )
    parser.add_argument(
        '--version', default='latest',
        help=(
            'Default version to use if none is specified'
        )
    )

        '--cache', default='.cache',
        help=(
            'Cache downloaded files that are not in --depot here'
        ),
    )
    parser.add_argument(
        '--credential-env',
        action='append',
        default=[],
        dest='credential_envs',
        help=(
            'Environment variable to be evaluated for login:password, '
            'or a pair of environment variables VAR1:VAR2 to be evaluated '
            'for login and password respectively'
        ),
    )
    parser.add_argument(
        '--credential-host',
        action='append',
        default=[],
        dest='credential_hosts',
        metavar='HOST',
        help=(
            'Use --credential-env when downloading from the given HOST'
            '(default: hostname of --images-uri)'
        ),
    )
    parser.add_argument(
        '--images-uri',
        default=DEFAULT_IMAGES_URI,
        metavar='URI',
        help=(
            'Download files from the given URI. '
            '"SUITE" will be replaced with the suite name.'
        ),
    )
        '--ssh-host', default='', metavar='HOST',
        help='Use ssh and rsync to download files from HOST',
    )
    parser.add_argument(
        '--ssh-path', default='', metavar='PATH',
        help=(
            'Use ssh and rsync to download files from PATH on HOST. '
            '"SUITE" will be replaced with the suite name.'
        ),
    parser.add_argument(
        '--depot', default='depot',
        help=(
            'Download runtimes into this existing directory'
        )
    )
    parser.add_argument(
        '--pressure-vessel', default='scout', metavar='NAME|PATH|DETAILS',
        help=(
            'Get pressure-vessel from the named runtime (default "scout"), '
            'or from a runtime version given as a JSON object, '
            'or from a given directory (use ./ to disambiguate if necessary).'
        )
    )
    parser.add_argument(
        '--include-archives', action='store_true', default=True,
        help=(
            'Provide the runtime as an archive to be unpacked '
            '[default]'
        )
    )
    parser.add_argument(
        '--no-include-archives', action='store_false', dest='include_archives',
        help=(
            'Do not provide the runtime as an archive to be unpacked'
        )
    )
    parser.add_argument(
        '--include-sdk', default=False, action='store_true',
        help='Include a corresponding SDK',
    )
    parser.add_argument(
        '--source-dir', default=HERE,
        help=(
            'Source directory for files to include in the depot'
        )
    )
    parser.add_argument(
        '--toolmanifest', default=False, action='store_true',
        help='Generate toolmanifest.vdf',
    )
    parser.add_argument(
        '--unpack-ld-library-path', metavar='PATH', default='',
        help=(
            'Get the steam-runtime.tar.xz from the same place as '
            'pressure-vessel and unpack it into the given PATH, '
            'for use in regression testing.'
        )
    )
    parser.add_argument(
        '--unpack-runtimes', action='store_true', default=False,
        help=(
            "Unpack the runtimes into the --depot, for use with "
            "pressure-vessel's tests/containers.py."
        )
    )
    parser.add_argument(
        '--no-unpack-runtimes', action='store_false', dest='unpack_runtimes',
        help=(
            "Don't unpack the runtimes into the --depot"
        )
    )
    parser.add_argument(
        '--unpack-source', metavar='PACKAGE', action='append', default=[],
        dest='unpack_sources',
        help=(
            'Download and unpack the given source package from each runtime '
            'if it exists, for use in regression testing. May be repeated.'
        )
    )
    parser.add_argument(
        '--unpack-sources-into', metavar='PATH', default='.',
        help=(
            'Unpack any source packages specified by --unpack-source '
            'into PATH/RUNTIME/SOURCE (default: ./RUNTIME/SOURCE).'
        )
    )
    parser.add_argument(
        '--versioned-directories', action='store_true', default=False,
        help=(
            'Include version number in unpacked runtime directories'
        )
    )
    parser.add_argument(
        '--no-versioned-directories', action='store_false',
        dest='versioned_directories',
        help=(
            'Do not include version number in unpacked runtime directories '
            '[default]'
        )
    )
    parser.add_argument(
        'runtimes',
        default=[],
        metavar='NAME[="DETAILS"]',
        nargs='*',
        help=(
            'Runtimes to download, in the form NAME or NAME="DETAILS". '
            'DETAILS is a JSON object containing something like '
            '{"path": "../prebuilt", "suite: "scout", "version": "latest", '
            '"architecture": "amd64,i386", "include_sdk": true}, or the '
            'path to a file with the same JSON object in. All JSON fields '
            'are optional.'
        ),
    )

    try:
        args = parser.parse_args()
        Main(**vars(args)).run()
    except InvocationError as e:
        parser.error(str(e))


if __name__ == '__main__':
    main()