from __future__ import annotations import hashlib import json import os from typing import Sequence from pre_commit import lang_base from pre_commit.prefix import Prefix from pre_commit.util import CalledProcessError from pre_commit.util import cmd_output_b ENVIRONMENT_DIR = 'docker' PRE_COMMIT_LABEL = 'PRE_COMMIT' get_default_version = lang_base.basic_get_default_version health_check = lang_base.basic_health_check in_env = lang_base.no_env # no special environment for docker def _is_in_docker() -> bool: try: with open('/proc/1/cgroup', 'rb') as f: return b'docker' in f.read() except FileNotFoundError: return False def _get_container_id() -> str: # It's assumed that we already check /proc/1/cgroup in _is_in_docker. The # cpuset cgroup controller existed since cgroups were introduced so this # way of getting the container ID is pretty reliable. with open('/proc/1/cgroup', 'rb') as f: for line in f.readlines(): if line.split(b':')[1] == b'cpuset': return os.path.basename(line.split(b':')[2]).strip().decode() raise RuntimeError('Failed to find the container ID in /proc/1/cgroup.') def _get_docker_path(path: str) -> str: if not _is_in_docker(): return path container_id = _get_container_id() try: _, out, _ = cmd_output_b('docker', 'inspect', container_id) except CalledProcessError: # self-container was not visible from here (perhaps docker-in-docker) return path container, = json.loads(out) for mount in container['Mounts']: src_path = mount['Source'] to_path = mount['Destination'] if os.path.commonpath((path, to_path)) == to_path: # So there is something in common, # and we can proceed remapping it return path.replace(to_path, src_path) # we're in Docker, but the path is not mounted, cannot really do anything, # so fall back to original path return path def md5(s: str) -> str: # pragma: win32 no cover return hashlib.md5(s.encode()).hexdigest() def docker_tag(prefix: Prefix) -> str: # pragma: win32 no cover md5sum = md5(os.path.basename(prefix.prefix_dir)).lower() return f'pre-commit-{md5sum}' def build_docker_image( prefix: Prefix, *, pull: bool, ) -> None: # pragma: win32 no cover cmd: tuple[str, ...] = ( 'docker', 'build', '--tag', docker_tag(prefix), '--label', PRE_COMMIT_LABEL, ) if pull: cmd += ('--pull',) # This must come last for old versions of docker. See #477 cmd += ('.',) lang_base.setup_cmd(prefix, cmd) def install_environment( prefix: Prefix, version: str, additional_dependencies: Sequence[str], ) -> None: # pragma: win32 no cover lang_base.assert_version_default('docker', version) lang_base.assert_no_additional_deps('docker', additional_dependencies) directory = lang_base.environment_dir(prefix, ENVIRONMENT_DIR, version) # Docker doesn't really have relevant disk environment, but pre-commit # still needs to cleanup its state files on failure build_docker_image(prefix, pull=True) os.mkdir(directory) def get_docker_user() -> tuple[str, ...]: # pragma: win32 no cover try: return ('-u', f'{os.getuid()}:{os.getgid()}') except AttributeError: return () def docker_cmd() -> tuple[str, ...]: # pragma: win32 no cover return ( 'docker', 'run', '--rm', *get_docker_user(), # https://docs.docker.com/engine/reference/commandline/run/#mount-volumes-from-container-volumes-from # The `Z` option tells Docker to label the content with a private # unshared label. Only the current container can use a private volume. '-v', f'{_get_docker_path(os.getcwd())}:/src:rw,Z', '--workdir', '/src', ) def run_hook( prefix: Prefix, entry: str, args: Sequence[str], file_args: Sequence[str], *, is_local: bool, require_serial: bool, color: bool, ) -> tuple[int, bytes]: # pragma: win32 no cover # Rebuild the docker image in case it has gone missing, as many people do # automated cleanup of docker images. build_docker_image(prefix, pull=False) entry_exe, *cmd_rest = lang_base.hook_cmd(entry, args) entry_tag = ('--entrypoint', entry_exe, docker_tag(prefix)) return lang_base.run_xargs( (*docker_cmd(), *entry_tag, *cmd_rest), file_args, require_serial=require_serial, color=color, )