majuna/app/terraform/terraform.py

import json
import subprocess  # nosec
from abc import abstractmethod
from typing import Any, Optional, Tuple

from app.terraform import BaseAutomation


class TerraformAutomation(BaseAutomation):
    """
    An abstract class to be extended by automation plugins using Terraform
    providers to deploy resources.
    """

    always_refresh: bool = False
    """
    Force refresh even when not a full run.
    """

    parallelism: int = 10
    """
    Default parallelism for remote API calls.
    """

    provider: str
    """
    Short name for the provider used by this module.
    """

    def automate(self, full: bool = False) -> Tuple[bool, str]:
        """
        Runs the Terraform automation module. The run will follow these steps:

        1. The :func:`tf_prehook` hook is run.
        2. Generate a Terraform configuration and write it to a single ``main.tf`` file in the working directory
           (see :func:`working_directory <app.terraform.BaseAutomation.working_directory>`).
        3. Run ``terraform init``.
        4. Run ``terraform apply``. This will only include a refresh if *full* is **True**. The apply will wait up
           to *lock_timeout* minutes for a lock to be released before failing. Up to *parallelism* requests will be
           sent to remote APIs concurrently.
        5. The :func:`tf_posthook` hook is run.
        6. The logs from the apply step are returned as a string.

        :param full: include a Terraform refresh in the automation module run
        :return: success status and Terraform apply logs
        """

        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        prehook_result = self.tf_prehook()  # pylint: disable=assignment-from-no-return
        self.tf_generate()
        self.tf_init()
        returncode, logs = self.tf_apply(self.working_dir, refresh=self.always_refresh or full)
        self.tf_posthook(prehook_result=prehook_result)
        return returncode == 0, logs

    def tf_apply(self, working_dir: str, *,
                 refresh: bool = True,
                 parallelism: Optional[int] = None,
                 lock_timeout: int = 15) -> Tuple[int, str]:
        if not parallelism:
            parallelism = self.parallelism
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        # The following subprocess call takes external input, but is providing
        # the argument list as an array such that argument injection would be
        # ineffective.
        tfcmd = subprocess.run(  # nosec
            ['terraform',
             'apply',
             '-auto-approve',
             '-json',
             f'-refresh={str(refresh).lower()}',
             f'-parallelism={str(parallelism)}',
             f'-lock-timeout={str(lock_timeout)}m',
             ],
            cwd=working_dir,
            stdout=subprocess.PIPE)
        return tfcmd.returncode, tfcmd.stdout.decode('utf-8')

    @abstractmethod
    def tf_generate(self) -> None:
        raise NotImplementedError()

    def tf_init(self, *, lock_timeout: int = 15) -> None:
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        # The init command does not support JSON output.
        # The following subprocess call takes external input, but is providing
        # the argument list as an array such that argument injection would be
        # ineffective.
        subprocess.run(  # nosec
            ['terraform',
             'init',
             f'-lock-timeout={str(lock_timeout)}m',
             ],
            cwd=self.working_dir)

    def tf_output(self) -> Any:
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        # The following subprocess call does not take any user input.
        tfcmd = subprocess.run(  # nosec
            ['terraform', 'output', '-json'],
            cwd=self.working_dir,
            stdout=subprocess.PIPE)
        return json.loads(tfcmd.stdout)

    def tf_plan(self, *,
                refresh: bool = True,
                parallelism: Optional[int] = None,
                lock_timeout: int = 15) -> Tuple[int, str]:
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        # The following subprocess call takes external input, but is providing
        # the argument list as an array such that argument injection would be
        # ineffective.
        tfcmd = subprocess.run(  # nosec
            ['terraform',
             'plan',
             '-json',
             f'-refresh={str(refresh).lower()}',
             f'-parallelism={str(parallelism)}',
             f'-lock-timeout={str(lock_timeout)}m',
             ],
            cwd=self.working_dir)
        return tfcmd.returncode, tfcmd.stdout.decode('utf-8')

    def tf_posthook(self, *, prehook_result: Any = None) -> None:
        """
        This hook function is called as part of normal automation, after the
        completion of :func:`tf_apply`.

        The default, if not overridden by a subclass, is to do nothing.

        :param prehook_result: the returned value of :func:`tf_prehook`
        :return: None
        """

    def tf_prehook(self) -> Optional[Any]:
        """
        This hook function is called as part of normal automation, before generating
        the terraform configuration file. The return value will be passed to
        :func:`tf_posthook` but is otherwise ignored.

        The default, if not overridden by a subclass, is to do nothing.

        :return: state that is useful to :func:`tf_posthook`, if required
        """

    def tf_show(self) -> Any:
        if not self.working_dir:
            raise RuntimeError("No working directory specified.")
        # This subprocess call doesn't take any user input.
        terraform = subprocess.run(  # nosec
            ['terraform', 'show', '-json'],
            cwd=self.working_dir,
            stdout=subprocess.PIPE)
        return json.loads(terraform.stdout)

    def tf_write(self, template: str, **kwargs: Any) -> None:
        self.tmpl_write("main.tf", template, **kwargs)