dulwich/dulwich/hooks.py

# hooks.py -- for dealing with git hooks
# Copyright (C) 2012-2013 Jelmer Vernooij and others.
#
# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as published by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#

"""Access to hooks."""

__all__ = [
    "CommitMsgShellHook",
    "Hook",
    "PostCommitShellHook",
    "PostReceiveShellHook",
    "PreCommitShellHook",
    "ShellHook",
]

import os
import subprocess
from collections.abc import Callable, Sequence
from typing import Any

from .errors import HookError


class Hook:
    """Generic hook object."""

    def execute(self, *args: Any) -> Any:  # noqa: ANN401
        """Execute the hook with the given args.

        Args:
          args: argument list to hook
        Raises:
          HookError: hook execution failure
        Returns:
          a hook may return a useful value
        """
        raise NotImplementedError(self.execute)


class ShellHook(Hook):
    """Hook by executable file.

    Implements standard githooks(5) [0]:

    [0] http://www.kernel.org/pub/software/scm/git/docs/githooks.html
    """

    def __init__(
        self,
        name: str,
        path: str,
        numparam: int,
        pre_exec_callback: Callable[..., Any] | None = None,
        post_exec_callback: Callable[..., Any] | None = None,
        cwd: str | None = None,
    ) -> None:
        """Setup shell hook definition.

        Args:
          name: name of hook for error messages
          path: absolute path to executable file
          numparam: number of requirements parameters
          pre_exec_callback: closure for setup before execution
            Defaults to None. Takes in the variable argument list from the
            execute functions and returns a modified argument list for the
            shell hook.
          post_exec_callback: closure for cleanup after execution
            Defaults to None. Takes in a boolean for hook success and the
            modified argument list and returns the final hook return value
            if applicable
          cwd: working directory to switch to when executing the hook
        """
        self.name = name
        self.filepath = path
        self.numparam = numparam

        self.pre_exec_callback = pre_exec_callback
        self.post_exec_callback = post_exec_callback

        self.cwd = cwd

    def execute(self, *args: Any) -> Any:  # noqa: ANN401
        """Execute the hook with given args."""
        if len(args) != self.numparam:
            raise HookError(
                f"Hook {self.name} executed with wrong number of args. Expected {self.numparam}. Saw {len(args)}. args: {args}"
            )

        if self.pre_exec_callback is not None:
            args = self.pre_exec_callback(*args)

        try:
            ret = subprocess.call(
                [os.path.relpath(self.filepath, self.cwd), *list(args)], cwd=self.cwd
            )
            if ret != 0:
                if self.post_exec_callback is not None:
                    self.post_exec_callback(0, *args)
                raise HookError(f"Hook {self.name} exited with non-zero status {ret}")
            if self.post_exec_callback is not None:
                return self.post_exec_callback(1, *args)
        except FileNotFoundError:  # no file. silent failure.
            if self.post_exec_callback is not None:
                self.post_exec_callback(0, *args)


class PreCommitShellHook(ShellHook):
    """pre-commit shell hook."""

    def __init__(self, cwd: str, controldir: str) -> None:
        """Initialize pre-commit hook.

        Args:
            cwd: Working directory for hook execution
            controldir: Path to the git control directory (.git)
        """
        filepath = os.path.join(controldir, "hooks", "pre-commit")

        ShellHook.__init__(self, "pre-commit", filepath, 0, cwd=cwd)


class PostCommitShellHook(ShellHook):
    """post-commit shell hook."""

    def __init__(self, controldir: str) -> None:
        """Initialize post-commit hook.

        Args:
            controldir: Path to the git control directory (.git)
        """
        filepath = os.path.join(controldir, "hooks", "post-commit")

        ShellHook.__init__(self, "post-commit", filepath, 0, cwd=controldir)


class CommitMsgShellHook(ShellHook):
    """commit-msg shell hook."""

    def __init__(self, controldir: str) -> None:
        """Initialize commit-msg hook.

        Args:
            controldir: Path to the git control directory (.git)
        """
        filepath = os.path.join(controldir, "hooks", "commit-msg")

        def prepare_msg(*args: bytes) -> tuple[str, ...]:
            import tempfile

            (fd, path) = tempfile.mkstemp()

            with os.fdopen(fd, "wb") as f:
                f.write(args[0])

            return (path,)

        def clean_msg(success: int, *args: str) -> bytes | None:
            if success:
                with open(args[0], "rb") as f:
                    new_msg = f.read()
                os.unlink(args[0])
                return new_msg
            os.unlink(args[0])
            return None

        ShellHook.__init__(
            self, "commit-msg", filepath, 1, prepare_msg, clean_msg, controldir
        )


class PostReceiveShellHook(ShellHook):
    """post-receive shell hook."""

    def __init__(self, controldir: str) -> None:
        """Initialize post-receive hook.

        Args:
            controldir: Path to the git control directory (.git)
        """
        self.controldir = controldir
        filepath = os.path.join(controldir, "hooks", "post-receive")
        ShellHook.__init__(self, "post-receive", path=filepath, numparam=0)

    def execute(
        self, client_refs: Sequence[tuple[bytes, bytes, bytes]]
    ) -> bytes | None:
        """Execute the post-receive hook.

        Args:
            client_refs: List of tuples containing (old_sha, new_sha, ref_name)
                        for each updated reference

        Returns:
            Output from the hook execution or None if hook doesn't exist

        Raises:
            HookError: If hook execution fails
        """
        # do nothing if the script doesn't exist
        if not os.path.exists(self.filepath):
            return None

        try:
            env = os.environ.copy()
            env["GIT_DIR"] = self.controldir

            p = subprocess.Popen(
                self.filepath,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env,
            )

            # client_refs is a list of (oldsha, newsha, ref)
            in_data = b"\n".join([b" ".join(ref) for ref in client_refs])

            out_data, err_data = p.communicate(in_data)

            if (p.returncode != 0) or err_data:
                err_msg = (
                    f"post-receive exit code: {p.returncode}\n"
                    f"stdout:\n{out_data.decode('utf-8', 'backslashreplace')}\n"
                    f"stderr:\n{err_data.decode('utf-8', 'backslashreplace')}"
                )
                raise HookError(err_msg)
            return out_data
        except OSError as err:
            raise HookError(repr(err)) from err