|
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
# Copyright 2015 The Meson development team
|
|
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
from .. import mlog
|
|
|
|
import contextlib
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import urllib.request
|
|
|
|
import urllib.error
|
|
|
|
import urllib.parse
|
|
|
|
import os
|
|
|
|
import hashlib
|
|
|
|
import shutil
|
|
|
|
import tempfile
|
|
|
|
import stat
|
|
|
|
import subprocess
|
|
|
|
import sys
|
|
|
|
import configparser
|
|
|
|
import time
|
|
|
|
import typing as T
|
|
|
|
import textwrap
|
|
|
|
import json
|
|
|
|
|
|
|
|
from base64 import b64encode
|
|
|
|
from netrc import netrc
|
|
|
|
from pathlib import Path, PurePath
|
|
|
|
from functools import lru_cache
|
|
|
|
|
|
|
|
from . import WrapMode
|
|
|
|
from .. import coredata
|
|
|
|
from ..mesonlib import quiet_git, GIT, ProgressBar, MesonException, windows_proof_rmtree, Popen_safe
|
|
|
|
from ..interpreterbase import FeatureNew
|
|
|
|
from ..interpreterbase import SubProject
|
|
|
|
from .. import mesonlib
|
|
|
|
|
|
|
|
if T.TYPE_CHECKING:
|
|
|
|
import http.client
|
|
|
|
from typing_extensions import Literal
|
|
|
|
|
|
|
|
Method = Literal['meson', 'cmake', 'cargo']
|
|
|
|
|
|
|
|
try:
|
|
|
|
# Importing is just done to check if SSL exists, so all warnings
|
|
|
|
# regarding 'imported but unused' can be safely ignored
|
|
|
|
import ssl # noqa
|
|
|
|
has_ssl = True
|
|
|
|
except ImportError:
|
|
|
|
has_ssl = False
|
|
|
|
|
|
|
|
REQ_TIMEOUT = 30.0
|
|
|
|
WHITELIST_SUBDOMAIN = 'wrapdb.mesonbuild.com'
|
|
|
|
|
|
|
|
ALL_TYPES = ['file', 'git', 'hg', 'svn']
|
|
|
|
|
|
|
|
PATCH = shutil.which('patch')
|
|
|
|
|
|
|
|
def whitelist_wrapdb(urlstr: str) -> urllib.parse.ParseResult:
|
|
|
|
""" raises WrapException if not whitelisted subdomain """
|
|
|
|
url = urllib.parse.urlparse(urlstr)
|
|
|
|
if not url.hostname:
|
|
|
|
raise WrapException(f'{urlstr} is not a valid URL')
|
|
|
|
if not url.hostname.endswith(WHITELIST_SUBDOMAIN):
|
|
|
|
raise WrapException(f'{urlstr} is not a whitelisted WrapDB URL')
|
|
|
|
if has_ssl and not url.scheme == 'https':
|
|
|
|
raise WrapException(f'WrapDB did not have expected SSL https url, instead got {urlstr}')
|
|
|
|
return url
|
|
|
|
|
|
|
|
def open_wrapdburl(urlstring: str, allow_insecure: bool = False, have_opt: bool = False) -> 'http.client.HTTPResponse':
|
|
|
|
if have_opt:
|
|
|
|
insecure_msg = '\n\n To allow connecting anyway, pass `--allow-insecure`.'
|
|
|
|
else:
|
|
|
|
insecure_msg = ''
|
|
|
|
|
|
|
|
url = whitelist_wrapdb(urlstring)
|
|
|
|
if has_ssl:
|
|
|
|
try:
|
|
|
|
return T.cast('http.client.HTTPResponse', urllib.request.urlopen(urllib.parse.urlunparse(url), timeout=REQ_TIMEOUT))
|
|
|
|
except urllib.error.URLError as excp:
|
|
|
|
msg = f'WrapDB connection failed to {urlstring} with error {excp}.'
|
|
|
|
if isinstance(excp.reason, ssl.SSLCertVerificationError):
|
|
|
|
if allow_insecure:
|
|
|
|
mlog.warning(f'{msg}\n\n Proceeding without authentication.')
|
|
|
|
else:
|
|
|
|
raise WrapException(f'{msg}{insecure_msg}')
|
|
|
|
else:
|
|
|
|
raise WrapException(msg)
|
|
|
|
elif not allow_insecure:
|
|
|
|
raise WrapException(f'SSL module not available in {sys.executable}: Cannot contact the WrapDB.{insecure_msg}')
|
|
|
|
else:
|
|
|
|
# following code is only for those without Python SSL
|
|
|
|
mlog.warning(f'SSL module not available in {sys.executable}: WrapDB traffic not authenticated.', once=True)
|
|
|
|
|
|
|
|
# If we got this far, allow_insecure was manually passed
|
|
|
|
nossl_url = url._replace(scheme='http')
|
|
|
|
try:
|
|
|
|
return T.cast('http.client.HTTPResponse', urllib.request.urlopen(urllib.parse.urlunparse(nossl_url), timeout=REQ_TIMEOUT))
|
|
|
|
except urllib.error.URLError as excp:
|
|
|
|
raise WrapException(f'WrapDB connection failed to {urlstring} with error {excp}')
|
|
|
|
|
|
|
|
def get_releases_data(allow_insecure: bool) -> bytes:
|
|
|
|
url = open_wrapdburl('https://wrapdb.mesonbuild.com/v2/releases.json', allow_insecure, True)
|
|
|
|
return url.read()
|
|
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
|
|
def get_releases(allow_insecure: bool) -> T.Dict[str, T.Any]:
|
|
|
|
data = get_releases_data(allow_insecure)
|
|
|
|
return T.cast('T.Dict[str, T.Any]', json.loads(data.decode()))
|
|
|
|
|
|
|
|
def update_wrap_file(wrapfile: str, name: str, new_version: str, new_revision: str, allow_insecure: bool) -> None:
|
|
|
|
url = open_wrapdburl(f'https://wrapdb.mesonbuild.com/v2/{name}_{new_version}-{new_revision}/{name}.wrap',
|
|
|
|
allow_insecure, True)
|
|
|
|
with open(wrapfile, 'wb') as f:
|
|
|
|
f.write(url.read())
|
|
|
|
|
|
|
|
def parse_patch_url(patch_url: str) -> T.Tuple[str, str]:
|
|
|
|
u = urllib.parse.urlparse(patch_url)
|
|
|
|
if u.netloc != 'wrapdb.mesonbuild.com':
|
|
|
|
raise WrapException(f'URL {patch_url} does not seems to be a wrapdb patch')
|
|
|
|
arr = u.path.strip('/').split('/')
|
|
|
|
if arr[0] == 'v1':
|
|
|
|
# e.g. https://wrapdb.mesonbuild.com/v1/projects/zlib/1.2.11/5/get_zip
|
|
|
|
return arr[-3], arr[-2]
|
|
|
|
elif arr[0] == 'v2':
|
|
|
|
# e.g. https://wrapdb.mesonbuild.com/v2/zlib_1.2.11-5/get_patch
|
|
|
|
tag = arr[-2]
|
|
|
|
_, version = tag.rsplit('_', 1)
|
|
|
|
version, revision = version.rsplit('-', 1)
|
|
|
|
return version, revision
|
|
|
|
else:
|
|
|
|
raise WrapException(f'Invalid wrapdb URL {patch_url}')
|
|
|
|
|
|
|
|
class WrapException(MesonException):
|
|
|
|
pass
|
|
|
|
|
|
|
|
class WrapNotFoundException(WrapException):
|
|
|
|
pass
|
|
|
|
|
|
|
|
class PackageDefinition:
|
|
|
|
def __init__(self, fname: str, subproject: str = ''):
|
|
|
|
self.filename = fname
|
|
|
|
self.subproject = SubProject(subproject)
|
|
|
|
self.type: T.Optional[str] = None
|
|
|
|
self.values: T.Dict[str, str] = {}
|
|
|
|
self.provided_deps: T.Dict[str, T.Optional[str]] = {}
|
|
|
|
self.provided_programs: T.List[str] = []
|
|
|
|
self.diff_files: T.List[Path] = []
|
|
|
|
self.basename = os.path.basename(fname)
|
|
|
|
self.has_wrap = self.basename.endswith('.wrap')
|
|
|
|
self.name = self.basename[:-5] if self.has_wrap else self.basename
|
|
|
|
# must be lowercase for consistency with dep=variable assignment
|
|
|
|
self.provided_deps[self.name.lower()] = None
|
|
|
|
# What the original file name was before redirection
|
|
|
|
self.original_filename = fname
|
|
|
|
self.redirected = False
|
|
|
|
if self.has_wrap:
|
|
|
|
self.parse_wrap()
|
|
|
|
with open(fname, 'r', encoding='utf-8') as file:
|
|
|
|
self.wrapfile_hash = hashlib.sha256(file.read().encode('utf-8')).hexdigest()
|
|
|
|
self.directory = self.values.get('directory', self.name)
|
|
|
|
if os.path.dirname(self.directory):
|
|
|
|
raise WrapException('Directory key must be a name and not a path')
|
|
|
|
if self.type and self.type not in ALL_TYPES:
|
|
|
|
raise WrapException(f'Unknown wrap type {self.type!r}')
|
|
|
|
self.filesdir = os.path.join(os.path.dirname(self.filename), 'packagefiles')
|
|
|
|
|
|
|
|
def parse_wrap(self) -> None:
|
|
|
|
try:
|
|
|
|
config = configparser.ConfigParser(interpolation=None)
|
|
|
|
config.read(self.filename, encoding='utf-8')
|
|
|
|
except configparser.Error as e:
|
|
|
|
raise WrapException(f'Failed to parse {self.basename}: {e!s}')
|
|
|
|
self.parse_wrap_section(config)
|
|
|
|
if self.type == 'redirect':
|
|
|
|
# [wrap-redirect] have a `filename` value pointing to the real wrap
|
|
|
|
# file we should parse instead. It must be relative to the current
|
|
|
|
# wrap file location and must be in the form foo/subprojects/bar.wrap.
|
|
|
|
dirname = Path(self.filename).parent
|
|
|
|
fname = Path(self.values['filename'])
|
|
|
|
for i, p in enumerate(fname.parts):
|
|
|
|
if i % 2 == 0:
|
|
|
|
if p == '..':
|
|
|
|
raise WrapException('wrap-redirect filename cannot contain ".."')
|
|
|
|
else:
|
|
|
|
if p != 'subprojects':
|
|
|
|
raise WrapException('wrap-redirect filename must be in the form foo/subprojects/bar.wrap')
|
|
|
|
if fname.suffix != '.wrap':
|
|
|
|
raise WrapException('wrap-redirect filename must be a .wrap file')
|
|
|
|
fname = dirname / fname
|
|
|
|
if not fname.is_file():
|
|
|
|
raise WrapException(f'wrap-redirect {fname} filename does not exist')
|
|
|
|
self.filename = str(fname)
|
|
|
|
self.parse_wrap()
|
|
|
|
self.redirected = True
|
|
|
|
else:
|
|
|
|
self.parse_provide_section(config)
|
|
|
|
if 'patch_directory' in self.values:
|
|
|
|
FeatureNew('Wrap files with patch_directory', '0.55.0').use(self.subproject)
|
|
|
|
for what in ['patch', 'source']:
|
|
|
|
if f'{what}_filename' in self.values and f'{what}_url' not in self.values:
|
|
|
|
FeatureNew(f'Local wrap patch files without {what}_url', '0.55.0').use(self.subproject)
|
|
|
|
|
|
|
|
def parse_wrap_section(self, config: configparser.ConfigParser) -> None:
|
|
|
|
if len(config.sections()) < 1:
|
|
|
|
raise WrapException(f'Missing sections in {self.basename}')
|
|
|
|
self.wrap_section = config.sections()[0]
|
|
|
|
if not self.wrap_section.startswith('wrap-'):
|
|
|
|
raise WrapException(f'{self.wrap_section!r} is not a valid first section in {self.basename}')
|
|
|
|
self.type = self.wrap_section[5:]
|
|
|
|
self.values = dict(config[self.wrap_section])
|
|
|
|
if 'diff_files' in self.values:
|
|
|
|
FeatureNew('Wrap files with diff_files', '0.63.0').use(self.subproject)
|
|
|
|
for s in self.values['diff_files'].split(','):
|
|
|
|
path = Path(s.strip())
|
|
|
|
if path.is_absolute():
|
|
|
|
raise WrapException('diff_files paths cannot be absolute')
|
|
|
|
if '..' in path.parts:
|
|
|
|
raise WrapException('diff_files paths cannot contain ".."')
|
|
|
|
self.diff_files.append(path)
|
|
|
|
|
|
|
|
def parse_provide_section(self, config: configparser.ConfigParser) -> None:
|
|
|
|
if config.has_section('provides'):
|
|
|
|
raise WrapException('Unexpected "[provides]" section, did you mean "[provide]"?')
|
|
|
|
if config.has_section('provide'):
|
|
|
|
for k, v in config['provide'].items():
|
|
|
|
if k == 'dependency_names':
|
|
|
|
# A comma separated list of dependency names that does not
|
|
|
|
# need a variable name; must be lowercase for consistency with
|
|
|
|
# dep=variable assignment
|
|
|
|
names_dict = {n.strip().lower(): None for n in v.split(',')}
|
|
|
|
self.provided_deps.update(names_dict)
|
|
|
|
continue
|
|
|
|
if k == 'program_names':
|
|
|
|
# A comma separated list of program names
|
|
|
|
names_list = [n.strip() for n in v.split(',')]
|
|
|
|
self.provided_programs += names_list
|
|
|
|
continue
|
|
|
|
if not v:
|
|
|
|
m = (f'Empty dependency variable name for {k!r} in {self.basename}. '
|
|
|
|
'If the subproject uses meson.override_dependency() '
|
|
|
|
'it can be added in the "dependency_names" special key.')
|
|
|
|
raise WrapException(m)
|
|
|
|
self.provided_deps[k] = v
|
|
|
|
|
|
|
|
def get(self, key: str) -> str:
|
|
|
|
try:
|
|
|
|
return self.values[key]
|
|
|
|
except KeyError:
|
|
|
|
raise WrapException(f'Missing key {key!r} in {self.basename}')
|
|
|
|
|
|
|
|
def get_hashfile(self, subproject_directory: str) -> str:
|
|
|
|
return os.path.join(subproject_directory, '.meson-subproject-wrap-hash.txt')
|
|
|
|
|
|
|
|
def update_hash_cache(self, subproject_directory: str) -> None:
|
|
|
|
if self.has_wrap:
|
|
|
|
with open(self.get_hashfile(subproject_directory), 'w', encoding='utf-8') as file:
|
|
|
|
file.write(self.wrapfile_hash + '\n')
|
|
|
|
|
|
|
|
def get_directory(subdir_root: str, packagename: str) -> str:
|
|
|
|
fname = os.path.join(subdir_root, packagename + '.wrap')
|
|
|
|
if os.path.isfile(fname):
|
|
|
|
wrap = PackageDefinition(fname)
|
|
|
|
return wrap.directory
|
|
|
|
return packagename
|
|
|
|
|
|
|
|
def verbose_git(cmd: T.List[str], workingdir: str, check: bool = False) -> bool:
|
|
|
|
'''
|
|
|
|
Wrapper to convert GitException to WrapException caught in interpreter.
|
|
|
|
'''
|
|
|
|
try:
|
|
|
|
return mesonlib.verbose_git(cmd, workingdir, check=check)
|
|
|
|
except mesonlib.GitException as e:
|
|
|
|
raise WrapException(str(e))
|
|
|
|
|
|
|
|
@dataclass(eq=False)
|
|
|
|
class Resolver:
|
|
|
|
source_dir: str
|
|
|
|
subdir: str
|
|
|
|
subproject: str = ''
|
|
|
|
wrap_mode: WrapMode = WrapMode.default
|
|
|
|
wrap_frontend: bool = False
|
|
|
|
allow_insecure: bool = False
|
|
|
|
silent: bool = False
|
|
|
|
|
|
|
|
def __post_init__(self) -> None:
|
|
|
|
self.subdir_root = os.path.join(self.source_dir, self.subdir)
|
|
|
|
self.cachedir = os.environ.get('MESON_PACKAGE_CACHE_DIR') or os.path.join(self.subdir_root, 'packagecache')
|
|
|
|
self.wraps: T.Dict[str, PackageDefinition] = {}
|
|
|
|
self.netrc: T.Optional[netrc] = None
|
|
|
|
self.provided_deps: T.Dict[str, PackageDefinition] = {}
|
|
|
|
self.provided_programs: T.Dict[str, PackageDefinition] = {}
|
|
|
|
self.wrapdb: T.Dict[str, T.Any] = {}
|
|
|
|
self.wrapdb_provided_deps: T.Dict[str, str] = {}
|
|
|
|
self.wrapdb_provided_programs: T.Dict[str, str] = {}
|
|
|
|
self.load_wraps()
|
|
|
|
self.load_netrc()
|
|
|
|
self.load_wrapdb()
|
|
|
|
|
|
|
|
def load_netrc(self) -> None:
|
|
|
|
try:
|
|
|
|
self.netrc = netrc()
|
|
|
|
except FileNotFoundError:
|
|
|
|
return
|
|
|
|
except Exception as e:
|
|
|
|
mlog.warning(f'failed to process netrc file: {e}.', fatal=False)
|
|
|
|
|
|
|
|
def load_wraps(self) -> None:
|
|
|
|
if not os.path.isdir(self.subdir_root):
|
|
|
|
return
|
|
|
|
root, dirs, files = next(os.walk(self.subdir_root))
|
|
|
|
ignore_dirs = {'packagecache', 'packagefiles'}
|
|
|
|
for i in files:
|
|
|
|
if not i.endswith('.wrap'):
|
|
|
|
continue
|
|
|
|
fname = os.path.join(self.subdir_root, i)
|
|
|
|
wrap = PackageDefinition(fname, self.subproject)
|
|
|
|
self.wraps[wrap.name] = wrap
|
|
|
|
ignore_dirs |= {wrap.directory, wrap.name}
|
|
|
|
# Add dummy package definition for directories not associated with a wrap file.
|
|
|
|
for i in dirs:
|
|
|
|
if i in ignore_dirs:
|
|
|
|
continue
|
|
|
|
fname = os.path.join(self.subdir_root, i)
|
|
|
|
wrap = PackageDefinition(fname, self.subproject)
|
|
|
|
self.wraps[wrap.name] = wrap
|
|
|
|
|
|
|
|
for wrap in self.wraps.values():
|
|
|
|
self.add_wrap(wrap)
|
|
|
|
|
|
|
|
def add_wrap(self, wrap: PackageDefinition) -> None:
|
|
|
|
for k in wrap.provided_deps.keys():
|
|
|
|
if k in self.provided_deps:
|
|
|
|
prev_wrap = self.provided_deps[k]
|
|
|
|
m = f'Multiple wrap files provide {k!r} dependency: {wrap.basename} and {prev_wrap.basename}'
|
|
|
|
raise WrapException(m)
|
|
|
|
self.provided_deps[k] = wrap
|
|
|
|
for k in wrap.provided_programs:
|
|
|
|
if k in self.provided_programs:
|
|
|
|
prev_wrap = self.provided_programs[k]
|
|
|
|
m = f'Multiple wrap files provide {k!r} program: {wrap.basename} and {prev_wrap.basename}'
|
|
|
|
raise WrapException(m)
|
|
|
|
self.provided_programs[k] = wrap
|
|
|
|
|
|
|
|
def load_wrapdb(self) -> None:
|
|
|
|
try:
|
|
|
|
with Path(self.subdir_root, 'wrapdb.json').open('r', encoding='utf-8') as f:
|
|
|
|
self.wrapdb = json.load(f)
|
|
|
|
except FileNotFoundError:
|
|
|
|
return
|
|
|
|
for name, info in self.wrapdb.items():
|
|
|
|
self.wrapdb_provided_deps.update({i: name for i in info.get('dependency_names', [])})
|
|
|
|
self.wrapdb_provided_programs.update({i: name for i in info.get('program_names', [])})
|
|
|
|
|
|
|
|
def get_from_wrapdb(self, subp_name: str) -> PackageDefinition:
|
|
|
|
info = self.wrapdb.get(subp_name)
|
|
|
|
if not info:
|
|
|
|
return None
|
|
|
|
self.check_can_download()
|
|
|
|
latest_version = info['versions'][0]
|
|
|
|
version, revision = latest_version.rsplit('-', 1)
|
|
|
|
url = urllib.request.urlopen(f'https://wrapdb.mesonbuild.com/v2/{subp_name}_{version}-{revision}/{subp_name}.wrap')
|
|
|
|
fname = Path(self.subdir_root, f'{subp_name}.wrap')
|
|
|
|
with fname.open('wb') as f:
|
|
|
|
f.write(url.read())
|
|
|
|
mlog.log(f'Installed {subp_name} version {version} revision {revision}')
|
|
|
|
wrap = PackageDefinition(str(fname))
|
|
|
|
self.wraps[wrap.name] = wrap
|
|
|
|
self.add_wrap(wrap)
|
|
|
|
return wrap
|
|
|
|
|
|
|
|
def merge_wraps(self, other_resolver: 'Resolver') -> None:
|
|
|
|
for k, v in other_resolver.wraps.items():
|
|
|
|
self.wraps.setdefault(k, v)
|
|
|
|
for k, v in other_resolver.provided_deps.items():
|
|
|
|
self.provided_deps.setdefault(k, v)
|
|
|
|
for k, v in other_resolver.provided_programs.items():
|
|
|
|
self.provided_programs.setdefault(k, v)
|
|
|
|
|
|
|
|
def find_dep_provider(self, packagename: str) -> T.Tuple[T.Optional[str], T.Optional[str]]:
|
|
|
|
# Python's ini parser converts all key values to lowercase.
|
|
|
|
# Thus the query name must also be in lower case.
|
|
|
|
packagename = packagename.lower()
|
|
|
|
wrap = self.provided_deps.get(packagename)
|
|
|
|
if wrap:
|
|
|
|
dep_var = wrap.provided_deps.get(packagename)
|
|
|
|
return wrap.name, dep_var
|
|
|
|
wrap_name = self.wrapdb_provided_deps.get(packagename)
|
|
|
|
return wrap_name, None
|
|
|
|
|
|
|
|
def get_varname(self, subp_name: str, depname: str) -> T.Optional[str]:
|
|
|
|
wrap = self.wraps.get(subp_name)
|
|
|
|
return wrap.provided_deps.get(depname) if wrap else None
|
|
|
|
|
|
|
|
def find_program_provider(self, names: T.List[str]) -> T.Optional[str]:
|
|
|
|
for name in names:
|
|
|
|
wrap = self.provided_programs.get(name)
|
|
|
|
if wrap:
|
|
|
|
return wrap.name
|
|
|
|
wrap_name = self.wrapdb_provided_programs.get(name)
|
|
|
|
if wrap_name:
|
|
|
|
return wrap_name
|
|
|
|
return None
|
|
|
|
|
|
|
|
def resolve(self, packagename: str, force_method: T.Optional[Method] = None) -> T.Tuple[str, Method]:
|
|
|
|
self.packagename = packagename
|
|
|
|
self.directory = packagename
|
|
|
|
self.wrap = self.wraps.get(packagename)
|
|
|
|
if not self.wrap:
|
|
|
|
self.wrap = self.get_from_wrapdb(packagename)
|
|
|
|
if not self.wrap:
|
|
|
|
m = f'Neither a subproject directory nor a {self.packagename}.wrap file was found.'
|
|
|
|
raise WrapNotFoundException(m)
|
|
|
|
self.directory = self.wrap.directory
|
|
|
|
|
|
|
|
if self.wrap.has_wrap:
|
|
|
|
# We have a .wrap file, use directory relative to the location of
|
|
|
|
# the wrap file if it exists, otherwise source code will be placed
|
|
|
|
# into main project's subproject_dir even if the wrap file comes
|
|
|
|
# from another subproject.
|
|
|
|
self.dirname = os.path.join(os.path.dirname(self.wrap.filename), self.wrap.directory)
|
|
|
|
if not os.path.exists(self.dirname):
|
|
|
|
self.dirname = os.path.join(self.subdir_root, self.directory)
|
|
|
|
# Check if the wrap comes from the main project.
|
|
|
|
main_fname = os.path.join(self.subdir_root, self.wrap.basename)
|
|
|
|
if self.wrap.filename != main_fname:
|
|
|
|
rel = os.path.relpath(self.wrap.filename, self.source_dir)
|
|
|
|
mlog.log('Using', mlog.bold(rel))
|
|
|
|
# Write a dummy wrap file in main project that redirect to the
|
|
|
|
# wrap we picked.
|
|
|
|
with open(main_fname, 'w', encoding='utf-8') as f:
|
|
|
|
f.write(textwrap.dedent(f'''\
|
|
|
|
[wrap-redirect]
|
|
|
|
filename = {PurePath(os.path.relpath(self.wrap.filename, self.subdir_root)).as_posix()}
|
|
|
|
'''))
|
|
|
|
else:
|
|
|
|
# No wrap file, it's a dummy package definition for an existing
|
|
|
|
# directory. Use the source code in place.
|
|
|
|
self.dirname = self.wrap.filename
|
|
|
|
rel_path = os.path.relpath(self.dirname, self.source_dir)
|
|
|
|
|
|
|
|
# Map each supported method to a file that must exist at the root of source tree.
|
|
|
|
methods_map: T.Dict[Method, str] = {
|
|
|
|
'meson': 'meson.build',
|
|
|
|
'cmake': 'CMakeLists.txt',
|
|
|
|
'cargo': 'Cargo.toml',
|
|
|
|
}
|
|
|
|
|
|
|
|
# Check if this wrap forces a specific method, use meson otherwise.
|
|
|
|
method = T.cast('T.Optional[Method]', self.wrap.values.get('method', force_method))
|
|
|
|
if method and method not in methods_map:
|
|
|
|
allowed_methods = ', '.join(methods_map.keys())
|
|
|
|
raise WrapException(f'Wrap method {method!r} is not supported, must be one of: {allowed_methods}')
|
|
|
|
if force_method and method != force_method:
|
|
|
|
raise WrapException(f'Wrap method is {method!r} but we are trying to configure it with {force_method}')
|
|
|
|
method = method or 'meson'
|
|
|
|
|
|
|
|
def has_buildfile() -> bool:
|
|
|
|
return os.path.exists(os.path.join(self.dirname, methods_map[method]))
|
|
|
|
|
|
|
|
# The directory is there and has meson.build? Great, use it.
|
|
|
|
if has_buildfile():
|
|
|
|
self.validate()
|
|
|
|
return rel_path, method
|
|
|
|
|
|
|
|
# Check if the subproject is a git submodule
|
|
|
|
self.resolve_git_submodule()
|
|
|
|
|
|
|
|
if os.path.exists(self.dirname):
|
|
|
|
if not os.path.isdir(self.dirname):
|
|
|
|
raise WrapException('Path already exists but is not a directory')
|
|
|
|
else:
|
|
|
|
# Check first if we have the extracted directory in our cache. This can
|
|
|
|
# happen for example when MESON_PACKAGE_CACHE_DIR=/usr/share/cargo/registry
|
|
|
|
# on distros that ships Rust source code.
|
|
|
|
# TODO: We don't currently clone git repositories into the cache
|
|
|
|
# directory, but we should to avoid cloning multiple times the same
|
|
|
|
# repository. In that case, we could do something smarter than
|
|
|
|
# copy_tree() here.
|
|
|
|
cached_directory = os.path.join(self.cachedir, self.directory)
|
|
|
|
if os.path.isdir(cached_directory):
|
|
|
|
self.copy_tree(cached_directory, self.dirname)
|
|
|
|
elif self.wrap.type == 'file':
|
|
|
|
self.get_file()
|
|
|
|
else:
|
|
|
|
self.check_can_download()
|
|
|
|
if self.wrap.type == 'git':
|
|
|
|
self.get_git()
|
|
|
|
elif self.wrap.type == "hg":
|
|
|
|
self.get_hg()
|
|
|
|
elif self.wrap.type == "svn":
|
|
|
|
self.get_svn()
|
|
|
|
else:
|
|
|
|
raise WrapException(f'Unknown wrap type {self.wrap.type!r}')
|
|
|
|
try:
|
|
|
|
self.apply_patch()
|
|
|
|
self.apply_diff_files()
|
|
|
|
except Exception:
|
|
|
|
windows_proof_rmtree(self.dirname)
|
|
|
|
raise
|
|
|
|
|
|
|
|
if not has_buildfile():
|
|
|
|
raise WrapException(f'Subproject exists but has no {methods_map[method]} file.')
|
|
|
|
|
|
|
|
# At this point, the subproject has been successfully resolved for the
|
|
|
|
# first time so save off the hash of the entire wrap file for future
|
|
|
|
# reference.
|
|
|
|
self.wrap.update_hash_cache(self.dirname)
|
|
|
|
return rel_path, method
|
|
|
|
|
|
|
|
def check_can_download(self) -> None:
|
|
|
|
# Don't download subproject data based on wrap file if requested.
|
|
|
|
# Git submodules are ok (see above)!
|
|
|
|
if self.wrap_mode is WrapMode.nodownload:
|
|
|
|
m = 'Automatic wrap-based subproject downloading is disabled'
|
|
|
|
raise WrapException(m)
|
|
|
|
|
|
|
|
def resolve_git_submodule(self) -> bool:
|
|
|
|
# Is git installed? If not, we're probably not in a git repository and
|
|
|
|
# definitely cannot try to conveniently set up a submodule.
|
|
|
|
if not GIT:
|
|
|
|
return False
|
|
|
|
# Does the directory exist? Even uninitialised submodules checkout an
|
|
|
|
# empty directory to work in
|
|
|
|
if not os.path.isdir(self.dirname):
|
|
|
|
return False
|
|
|
|
# Are we in a git repository?
|
|
|
|
ret, out = quiet_git(['rev-parse'], Path(self.dirname).parent)
|
|
|
|
if not ret:
|
|
|
|
return False
|
|
|
|
# Is `dirname` a submodule?
|
|
|
|
ret, out = quiet_git(['submodule', 'status', '.'], self.dirname)
|
|
|
|
if not ret:
|
|
|
|
return False
|
|
|
|
# Submodule has not been added, add it
|
|
|
|
if out.startswith('+'):
|
|
|
|
mlog.warning('git submodule might be out of date')
|
|
|
|
return True
|
|
|
|
elif out.startswith('U'):
|
|
|
|
raise WrapException('git submodule has merge conflicts')
|
|
|
|
# Submodule exists, but is deinitialized or wasn't initialized
|
|
|
|
elif out.startswith('-'):
|
|
|
|
if verbose_git(['submodule', 'update', '--init', '.'], self.dirname):
|
|
|
|
return True
|
|
|
|
raise WrapException('git submodule failed to init')
|
|
|
|
# Submodule looks fine, but maybe it wasn't populated properly. Do a checkout.
|
|
|
|
elif out.startswith(' '):
|
|
|
|
verbose_git(['submodule', 'update', '.'], self.dirname)
|
|
|
|
verbose_git(['checkout', '.'], self.dirname)
|
|
|
|
# Even if checkout failed, try building it anyway and let the user
|
|
|
|
# handle any problems manually.
|
|
|
|
return True
|
|
|
|
elif out == '':
|
|
|
|
# It is not a submodule, just a folder that exists in the main repository.
|
|
|
|
return False
|
|
|
|
raise WrapException(f'Unknown git submodule output: {out!r}')
|
|
|
|
|
|
|
|
def get_file(self) -> None:
|
|
|
|
path = self.get_file_internal('source')
|
|
|
|
extract_dir = self.subdir_root
|
|
|
|
# Some upstreams ship packages that do not have a leading directory.
|
|
|
|
# Create one for them.
|
|
|
|
if 'lead_directory_missing' in self.wrap.values:
|
|
|
|
os.mkdir(self.dirname)
|
|
|
|
extract_dir = self.dirname
|
wrap: catch failure to extract tarball and raise WrapException
This most likely happens when the source archive has files which take
advantage of case sensitivity, and someone is unfortunate enough to have
to work with broken operating systems that do not have the capacity to
use modern technology, like the 1970s invention of case sensitive
filesystems.
For example, Windows and macOS both have retrocomputing environments,
where case sensitive abilities were carefully removed from modern
filesystems in order to share the delights of classical computing with
the masses.
On such systems, innocent tarballs fail to extract with:
```
OSError: [Errno 22] Invalid argument: 'C:\path\to\file'
```
thereby demonstrating Microsoft's unbounded and enthusiastic love for
users (who are, universally, retrocomputing fans).
n.b. there is, begrudgingly, a secret option for crazy people who hate
retrocomputing to enable modern 1970s computing on Windows systems.
Naturally, no one with any sense would ever use this, and it is rightly
left in its sensible default in for example Github Actions.
Fixes #12344
1 year ago
|
|
|
try:
|
|
|
|
shutil.unpack_archive(path, extract_dir)
|
|
|
|
except OSError as e:
|
|
|
|
raise WrapException(f'failed to unpack archive with error: {str(e)}') from e
|
|
|
|
|
|
|
|
def get_git(self) -> None:
|
|
|
|
if not GIT:
|
|
|
|
raise WrapException(f'Git program not found, cannot download {self.packagename}.wrap via git.')
|
|
|
|
revno = self.wrap.get('revision')
|
|
|
|
checkout_cmd = ['-c', 'advice.detachedHead=false', 'checkout', revno, '--']
|
|
|
|
is_shallow = False
|
|
|
|
depth_option: T.List[str] = []
|
|
|
|
if self.wrap.values.get('depth', '') != '':
|
|
|
|
is_shallow = True
|
|
|
|
depth_option = ['--depth', self.wrap.values.get('depth')]
|
|
|
|
# for some reason git only allows commit ids to be shallowly fetched by fetch not with clone
|
|
|
|
if is_shallow and self.is_git_full_commit_id(revno):
|
|
|
|
# git doesn't support directly cloning shallowly for commits,
|
|
|
|
# so we follow https://stackoverflow.com/a/43136160
|
|
|
|
verbose_git(['-c', 'init.defaultBranch=meson-dummy-branch', 'init', self.directory], self.subdir_root, check=True)
|
|
|
|
verbose_git(['remote', 'add', 'origin', self.wrap.get('url')], self.dirname, check=True)
|
|
|
|
revno = self.wrap.get('revision')
|
|
|
|
verbose_git(['fetch', *depth_option, 'origin', revno], self.dirname, check=True)
|
|
|
|
verbose_git(checkout_cmd, self.dirname, check=True)
|
|
|
|
else:
|
|
|
|
if not is_shallow:
|
|
|
|
verbose_git(['clone', self.wrap.get('url'), self.directory], self.subdir_root, check=True)
|
|
|
|
if revno.lower() != 'head':
|
|
|
|
if not verbose_git(checkout_cmd, self.dirname):
|
|
|
|
verbose_git(['fetch', self.wrap.get('url'), revno], self.dirname, check=True)
|
|
|
|
verbose_git(checkout_cmd, self.dirname, check=True)
|
|
|
|
else:
|
|
|
|
args = ['-c', 'advice.detachedHead=false', 'clone', *depth_option]
|
|
|
|
if revno.lower() != 'head':
|
|
|
|
args += ['--branch', revno]
|
|
|
|
args += [self.wrap.get('url'), self.directory]
|
|
|
|
verbose_git(args, self.subdir_root, check=True)
|
|
|
|
if self.wrap.values.get('clone-recursive', '').lower() == 'true':
|
|
|
|
verbose_git(['submodule', 'update', '--init', '--checkout', '--recursive', *depth_option],
|
|
|
|
self.dirname, check=True)
|
|
|
|
push_url = self.wrap.values.get('push-url')
|
|
|
|
if push_url:
|
|
|
|
verbose_git(['remote', 'set-url', '--push', 'origin', push_url], self.dirname, check=True)
|
|
|
|
|
|
|
|
def validate(self) -> None:
|
|
|
|
# This check is only for subprojects with wraps.
|
|
|
|
if not self.wrap.has_wrap:
|
|
|
|
return
|
|
|
|
|
|
|
|
# Retrieve original hash, if it exists.
|
|
|
|
hashfile = self.wrap.get_hashfile(self.dirname)
|
|
|
|
if os.path.isfile(hashfile):
|
|
|
|
with open(hashfile, 'r', encoding='utf-8') as file:
|
|
|
|
expected_hash = file.read().strip()
|
|
|
|
else:
|
|
|
|
# If stored hash doesn't exist then don't warn.
|
|
|
|
return
|
|
|
|
|
|
|
|
actual_hash = self.wrap.wrapfile_hash
|
|
|
|
|
|
|
|
# Compare hashes and warn the user if they don't match.
|
|
|
|
if expected_hash != actual_hash:
|
|
|
|
mlog.warning(f'Subproject {self.wrap.name}\'s revision may be out of date; its wrap file has changed since it was first configured')
|
|
|
|
|
|
|
|
def is_git_full_commit_id(self, revno: str) -> bool:
|
|
|
|
result = False
|
|
|
|
if len(revno) in {40, 64}: # 40 for sha1, 64 for upcoming sha256
|
|
|
|
result = all(ch in '0123456789AaBbCcDdEeFf' for ch in revno)
|
|
|
|
return result
|
|
|
|
|
|
|
|
def get_hg(self) -> None:
|
|
|
|
revno = self.wrap.get('revision')
|
|
|
|
hg = shutil.which('hg')
|
|
|
|
if not hg:
|
|
|
|
raise WrapException('Mercurial program not found.')
|
|
|
|
subprocess.check_call([hg, 'clone', self.wrap.get('url'),
|
|
|
|
self.directory], cwd=self.subdir_root)
|
|
|
|
if revno.lower() != 'tip':
|
|
|
|
subprocess.check_call([hg, 'checkout', revno],
|
|
|
|
cwd=self.dirname)
|
|
|
|
|
|
|
|
def get_svn(self) -> None:
|
|
|
|
revno = self.wrap.get('revision')
|
|
|
|
svn = shutil.which('svn')
|
|
|
|
if not svn:
|
|
|
|
raise WrapException('SVN program not found.')
|
|
|
|
subprocess.check_call([svn, 'checkout', '-r', revno, self.wrap.get('url'),
|
|
|
|
self.directory], cwd=self.subdir_root)
|
|
|
|
|
|
|
|
def get_netrc_credentials(self, netloc: str) -> T.Optional[T.Tuple[str, str]]:
|
|
|
|
if self.netrc is None or netloc not in self.netrc.hosts:
|
|
|
|
return None
|
|
|
|
|
|
|
|
login, account, password = self.netrc.authenticators(netloc)
|
|
|
|
if account is not None:
|
|
|
|
login = account
|
|
|
|
|
|
|
|
return login, password
|
|
|
|
|
|
|
|
def get_data(self, urlstring: str) -> T.Tuple[str, str]:
|
|
|
|
blocksize = 10 * 1024
|
|
|
|
h = hashlib.sha256()
|
|
|
|
tmpfile = tempfile.NamedTemporaryFile(mode='wb', dir=self.cachedir, delete=False)
|
|
|
|
url = urllib.parse.urlparse(urlstring)
|
|
|
|
if url.hostname and url.hostname.endswith(WHITELIST_SUBDOMAIN):
|
|
|
|
resp = open_wrapdburl(urlstring, allow_insecure=self.allow_insecure, have_opt=self.wrap_frontend)
|
|
|
|
elif WHITELIST_SUBDOMAIN in urlstring:
|
|
|
|
raise WrapException(f'{urlstring} may be a WrapDB-impersonating URL')
|
|
|
|
else:
|
|
|
|
headers = {'User-Agent': f'mesonbuild/{coredata.version}'}
|
|
|
|
creds = self.get_netrc_credentials(url.netloc)
|
|
|
|
|
|
|
|
if creds is not None and '@' not in url.netloc:
|
|
|
|
login, password = creds
|
|
|
|
if url.scheme == 'https':
|
|
|
|
enc_creds = b64encode(f'{login}:{password}'.encode()).decode()
|
|
|
|
headers.update({'Authorization': f'Basic {enc_creds}'})
|
|
|
|
elif url.scheme == 'ftp':
|
|
|
|
urlstring = urllib.parse.urlunparse(url._replace(netloc=f'{login}:{password}@{url.netloc}'))
|
|
|
|
else:
|
|
|
|
mlog.warning('Meson is not going to use netrc credentials for protocols other than https/ftp',
|
|
|
|
fatal=False)
|
|
|
|
|
|
|
|
try:
|
|
|
|
req = urllib.request.Request(urlstring, headers=headers)
|
|
|
|
resp = urllib.request.urlopen(req, timeout=REQ_TIMEOUT)
|
|
|
|
except urllib.error.URLError as e:
|
|
|
|
mlog.log(str(e))
|
|
|
|
raise WrapException(f'could not get {urlstring} is the internet available?')
|
|
|
|
with contextlib.closing(resp) as resp, tmpfile as tmpfile:
|
|
|
|
try:
|
|
|
|
dlsize = int(resp.info()['Content-Length'])
|
|
|
|
except TypeError:
|
|
|
|
dlsize = None
|
|
|
|
if dlsize is None:
|
|
|
|
print('Downloading file of unknown size.')
|
|
|
|
while True:
|
|
|
|
block = resp.read(blocksize)
|
|
|
|
if block == b'':
|
|
|
|
break
|
|
|
|
h.update(block)
|
|
|
|
tmpfile.write(block)
|
|
|
|
hashvalue = h.hexdigest()
|
|
|
|
return hashvalue, tmpfile.name
|
|
|
|
sys.stdout.flush()
|
|
|
|
progress_bar = ProgressBar(bar_type='download', total=dlsize,
|
|
|
|
desc='Downloading',
|
|
|
|
disable=(self.silent or None))
|
|
|
|
while True:
|
|
|
|
block = resp.read(blocksize)
|
|
|
|
if block == b'':
|
|
|
|
break
|
|
|
|
h.update(block)
|
|
|
|
tmpfile.write(block)
|
|
|
|
progress_bar.update(len(block))
|
|
|
|
progress_bar.close()
|
|
|
|
hashvalue = h.hexdigest()
|
|
|
|
return hashvalue, tmpfile.name
|
|
|
|
|
|
|
|
def check_hash(self, what: str, path: str, hash_required: bool = True) -> None:
|
|
|
|
if what + '_hash' not in self.wrap.values and not hash_required:
|
|
|
|
return
|
|
|
|
expected = self.wrap.get(what + '_hash').lower()
|
|
|
|
h = hashlib.sha256()
|
|
|
|
with open(path, 'rb') as f:
|
|
|
|
h.update(f.read())
|
|
|
|
dhash = h.hexdigest()
|
|
|
|
if dhash != expected:
|
|
|
|
raise WrapException(f'Incorrect hash for {what}:\n {expected} expected\n {dhash} actual.')
|
|
|
|
|
|
|
|
def get_data_with_backoff(self, urlstring: str) -> T.Tuple[str, str]:
|
|
|
|
delays = [1, 2, 4, 8, 16]
|
|
|
|
for d in delays:
|
|
|
|
try:
|
|
|
|
return self.get_data(urlstring)
|
|
|
|
except Exception as e:
|
|
|
|
mlog.warning(f'failed to download with error: {e}. Trying after a delay...', fatal=False)
|
|
|
|
time.sleep(d)
|
|
|
|
return self.get_data(urlstring)
|
|
|
|
|
|
|
|
def download(self, what: str, ofname: str, fallback: bool = False) -> None:
|
|
|
|
self.check_can_download()
|
|
|
|
srcurl = self.wrap.get(what + ('_fallback_url' if fallback else '_url'))
|
|
|
|
mlog.log('Downloading', mlog.bold(self.packagename), what, 'from', mlog.bold(srcurl))
|
|
|
|
try:
|
|
|
|
dhash, tmpfile = self.get_data_with_backoff(srcurl)
|
|
|
|
expected = self.wrap.get(what + '_hash').lower()
|
|
|
|
if dhash != expected:
|
|
|
|
os.remove(tmpfile)
|
|
|
|
raise WrapException(f'Incorrect hash for {what}:\n {expected} expected\n {dhash} actual.')
|
|
|
|
except WrapException:
|
|
|
|
if not fallback:
|
|
|
|
if what + '_fallback_url' in self.wrap.values:
|
|
|
|
return self.download(what, ofname, fallback=True)
|
|
|
|
mlog.log('A fallback URL could be specified using',
|
|
|
|
mlog.bold(what + '_fallback_url'), 'key in the wrap file')
|
|
|
|
raise
|
|
|
|
os.rename(tmpfile, ofname)
|
|
|
|
|
|
|
|
def get_file_internal(self, what: str) -> str:
|
|
|
|
filename = self.wrap.get(what + '_filename')
|
|
|
|
if what + '_url' in self.wrap.values:
|
|
|
|
cache_path = os.path.join(self.cachedir, filename)
|
|
|
|
|
|
|
|
if os.path.exists(cache_path):
|
|
|
|
self.check_hash(what, cache_path)
|
|
|
|
mlog.log('Using', mlog.bold(self.packagename), what, 'from cache.')
|
|
|
|
return cache_path
|
|
|
|
|
|
|
|
os.makedirs(self.cachedir, exist_ok=True)
|
|
|
|
self.download(what, cache_path)
|
|
|
|
return cache_path
|
|
|
|
else:
|
|
|
|
path = Path(self.wrap.filesdir) / filename
|
|
|
|
|
|
|
|
if not path.exists():
|
|
|
|
raise WrapException(f'File "{path}" does not exist')
|
|
|
|
self.check_hash(what, path.as_posix(), hash_required=False)
|
|
|
|
|
|
|
|
return path.as_posix()
|
|
|
|
|
|
|
|
def apply_patch(self) -> None:
|
|
|
|
if 'patch_filename' in self.wrap.values and 'patch_directory' in self.wrap.values:
|
|
|
|
m = f'Wrap file {self.wrap.basename!r} must not have both "patch_filename" and "patch_directory"'
|
|
|
|
raise WrapException(m)
|
|
|
|
if 'patch_filename' in self.wrap.values:
|
|
|
|
path = self.get_file_internal('patch')
|
|
|
|
try:
|
|
|
|
shutil.unpack_archive(path, self.subdir_root)
|
|
|
|
except Exception:
|
|
|
|
with tempfile.TemporaryDirectory() as workdir:
|
|
|
|
shutil.unpack_archive(path, workdir)
|
|
|
|
self.copy_tree(workdir, self.subdir_root)
|
|
|
|
elif 'patch_directory' in self.wrap.values:
|
|
|
|
patch_dir = self.wrap.values['patch_directory']
|
|
|
|
src_dir = os.path.join(self.wrap.filesdir, patch_dir)
|
|
|
|
if not os.path.isdir(src_dir):
|
|
|
|
raise WrapException(f'patch directory does not exist: {patch_dir}')
|
|
|
|
self.copy_tree(src_dir, self.dirname)
|
|
|
|
|
|
|
|
def apply_diff_files(self) -> None:
|
|
|
|
for filename in self.wrap.diff_files:
|
|
|
|
mlog.log(f'Applying diff file "{filename}"')
|
|
|
|
path = Path(self.wrap.filesdir) / filename
|
|
|
|
if not path.exists():
|
|
|
|
raise WrapException(f'Diff file "{path}" does not exist')
|
|
|
|
relpath = os.path.relpath(str(path), self.dirname)
|
|
|
|
if GIT:
|
|
|
|
# Git is more likely to be available on Windows and more likely
|
|
|
|
# to apply correctly assuming patches are often generated by git.
|
|
|
|
# See https://github.com/mesonbuild/meson/issues/12092.
|
|
|
|
# The `--work-tree` is necessary in case we're inside a
|
|
|
|
# Git repository: by default, Git will try to apply the patch to
|
|
|
|
# the repository root.
|
|
|
|
cmd = [GIT, '--work-tree', '.', 'apply', '-p1', relpath]
|
|
|
|
elif PATCH:
|
|
|
|
# Always pass a POSIX path to patch, because on Windows it's MSYS
|
|
|
|
cmd = [PATCH, '-f', '-p1', '-i', str(Path(relpath).as_posix())]
|
|
|
|
else:
|
|
|
|
raise WrapException('Missing "patch" or "git" commands to apply diff files')
|
|
|
|
|
|
|
|
p, out, _ = Popen_safe(cmd, cwd=self.dirname, stderr=subprocess.STDOUT)
|
|
|
|
if p.returncode != 0:
|
|
|
|
mlog.log(out.strip())
|
|
|
|
raise WrapException(f'Failed to apply diff file "{filename}"')
|
|
|
|
|
|
|
|
def copy_tree(self, root_src_dir: str, root_dst_dir: str) -> None:
|
|
|
|
"""
|
|
|
|
Copy directory tree. Overwrites also read only files.
|
|
|
|
"""
|
|
|
|
for src_dir, _, files in os.walk(root_src_dir):
|
|
|
|
dst_dir = src_dir.replace(root_src_dir, root_dst_dir, 1)
|
|
|
|
if not os.path.exists(dst_dir):
|
|
|
|
os.makedirs(dst_dir)
|
|
|
|
for file_ in files:
|
|
|
|
src_file = os.path.join(src_dir, file_)
|
|
|
|
dst_file = os.path.join(dst_dir, file_)
|
|
|
|
if os.path.exists(dst_file):
|
|
|
|
try:
|
|
|
|
os.remove(dst_file)
|
|
|
|
except PermissionError:
|
|
|
|
os.chmod(dst_file, stat.S_IWUSR)
|
|
|
|
os.remove(dst_file)
|
|
|
|
shutil.copy2(src_file, dst_dir)
|