Newer
Older
with suppress(FileNotFoundError):
logger.info('Removing %r', dest)
shutil.rmtree(dest)
subprocess.run(
[
'dpkg-source',
'-x',
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
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
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)))
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
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'
)
)
parser.add_argument(
'--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.'
),
)
parser.add_argument(
'--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-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(
'--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(
'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.'