Enable support for Vault secrets in Manifester

Makefile

@@ -0,0 +1,8 @@
+vault-login:
+	@scripts/vault_login.py --login
+
+vault-logout:
+	@scripts/vault_login.py --logout
+
+vault-status:
+	@scripts/vault_login.py --status

manifester/commands.py

@@ -60,8 +60,13 @@ def delete(allocations, all_, remove_manifest_file):
                 uuid=allocation.get("uuid")
             )
             if remove_manifest_file:
+                manifester_directory = (
+                    Path(os.environ["MANIFESTER_DIRECTORY"]).resolve()
+                    if "MANIFESTER_DIRECTORY" in os.environ
+                    else Path()
+                )
                 Path(
-                    f"{os.environ['MANIFESTER_DIRECTORY']}/manifests/{allocation.get('name')}_manifest.zip"
+                    f"{manifester_directory}/manifests/{allocation.get('name')}_manifest.zip"
                 ).unlink()
 
 

manifester/helpers.py

@@ -1,15 +1,24 @@
 """Defines helper functions used by Manifester."""
 from collections import UserDict
+import json
+import os
 from pathlib import Path
 import random
+import re
+import subprocess
+import sys
 import time
 
 from logzero import logger
 from requests import HTTPError
 import yaml
 
+from manifester.logger import setup_logzero
 from manifester.settings import settings
 
+setup_logzero(level="info")
+
+
 RESULTS_LIMIT = 10000
 
 
@@ -226,3 +235,139 @@ def __getitem__(self, key):
     def __call__(self, *args, **kwargs):
         """Allow MockStub to be used like a function."""
         return self
+
+
+class InvalidVaultURLForOIDC(Exception):
+    """Raised if the vault doesn't allow OIDC login."""
+
+
+class Vault:
+    """Helper class for retrieving secrets from HashiCorp Vault."""
+
+    HELP_TEXT = (
+        "The Vault CLI in not installed on this system."
+        "Please follow https://learn.hashicorp.com/tutorials/vault/getting-started-install to "
+        "install the Vault CLI."
+    )
+
+    def __init__(self, env_file=".env"):
+        manifester_directory = Path()
+
+        if "MANIFESTER_DIRECTORY" in os.environ:
+            envar_location = Path(os.environ["MANIFESTER_DIRECTORY"])
+            if envar_location.is_dir():
+                manifester_directory = envar_location
+        self.env_path = manifester_directory.joinpath(env_file)
+        self.envdata = None
+        self.vault_enabled = None
+
+    def setup(self):
+        """Read environment variables from .env."""
+        if self.env_path.exists():
+            self.envdata = self.env_path.read_text()
+            is_enabled = re.findall("^(?:.*\n)*VAULT_ENABLED_FOR_DYNACONF=(.*)", self.envdata)
+            if is_enabled:
+                self.vault_enabled = is_enabled[0]
+            self.export_vault_addr()
+
+    def teardown(self):
+        """Remove VAULT_ADDR environment variable if present."""
+        if os.environ.get("VAULT_ADDR") is not None:
+            del os.environ["VAULT_ADDR"]
+
+    def export_vault_addr(self):
+        """Set the URL of the Vault server and ensure that the URL is not localhost."""
+        vaulturl = re.findall("VAULT_URL_FOR_DYNACONF=(.*)", self.envdata)[0]
+
+        # Set Vault CLI Env Var
+        os.environ["VAULT_ADDR"] = vaulturl
+
+        # Dynaconf Vault Env Vars
+        if (
+            self.vault_enabled
+            and self.vault_enabled in ["True", "true"]
+            and "localhost:8200" in vaulturl
+        ):
+            raise InvalidVaultURLForOIDC(
+                f"{vaulturl} does not support OIDC login."
+                "Please set the correct vault URL vault the .env file."
+            )
+
+    def exec_vault_command(self, command: str, **kwargs):
+        """Wrap Vault CLI commands for execution.
+
+        :param comamnd str: The vault CLI command
+        :param kwargs dict: Arguments to the subprocess run command to customize the run behavior
+        """
+        COMMAND_NOT_FOUND_EXIT_CODE = 127
+        vcommand = subprocess.run(command.split(), capture_output=True, **kwargs)
+        if vcommand.returncode != 0:
+            verror = str(vcommand.stderr)
+            if vcommand.returncode == COMMAND_NOT_FOUND_EXIT_CODE:
+                logger.error(f"Error! {self.HELP_TEXT}")
+                sys.exit(1)
+            if vcommand.stderr:
+                if "Error revoking token" in verror:
+                    logger.info("Token is already revoked")
+                elif "Error looking up token" in verror:
+                    logger.info("Vault is not logged in")
+                else:
+                    logger.error(f"Error: {verror}")
+        return vcommand
+
+    def login(self, **kwargs):
+        """Authenticate to Vault server and add auth token to .env file."""
+        if (
+            self.vault_enabled
+            and self.vault_enabled in ["True", "true"]
+            and "VAULT_SECRET_ID_FOR_DYNACONF" not in os.environ
+            and self.status(**kwargs).returncode != 0
+        ):
+            logger.info(
+                "Warning: A browser tab will open for Vault OIDC login. "
+                "Please close the tab once the sign-in is complete"
+            )
+            if (
+                self.exec_vault_command(command="vault login -method=oidc", **kwargs).returncode
+                == 0
+            ):
+                self.exec_vault_command(command="vault token renew -i 10h", **kwargs)
+                logger.info("Success! Vault OIDC Logged-In and extended for 10 hours!")
+            # Fetch token
+            token = self.exec_vault_command("vault token lookup --format json").stdout
+            token = json.loads(str(token.decode("UTF-8")))["data"]["id"]
+            # Set new token in .env file
+            _envdata = re.sub(
+                ".*VAULT_TOKEN_FOR_DYNACONF=.*",
+                f"VAULT_TOKEN_FOR_DYNACONF={token}",
+                self.envdata,
+            )
+            self.env_path.write_text(_envdata)
+            logger.info("New OIDC token succesfully added to .env file")
+
+    def logout(self):
+        """Revoke Vault auth token and remove it from .env file."""
+        # Teardown - Setting dummy token in env file
+        _envdata = re.sub(
+            ".*VAULT_TOKEN_FOR_DYNACONF=.*", "# VAULT_TOKEN_FOR_DYNACONF=myroot", self.envdata
+        )
+        self.env_path.write_text(_envdata)
+        vstatus = self.exec_vault_command("vault token revoke -self")
+        if vstatus.returncode == 0:
+            logger.info("OIDC token successfully removed from .env file")
+
+    def status(self, **kwargs):
+        """Check status of Vault auth token."""
+        vstatus = self.exec_vault_command("vault token lookup", **kwargs)
+        if vstatus.returncode == 0:
+            logger.info(str(vstatus.stdout.decode("UTF-8")))
+        return vstatus
+
+    def __enter__(self):
+        """Set up Vault context manager."""
+        self.setup()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Tear down Vault context manager."""
+        self.teardown()

manifester/manifester.py

@@ -21,8 +21,6 @@
 from manifester.logger import setup_logzero
 from manifester.settings import settings
 
-setup_logzero(level=settings.get("log_level", "info"))
-
 
 class Manifester:
     """Main Manifester class responsible for generating a manifest from the provided settings."""
@@ -35,6 +33,7 @@ def __init__(
         proxies=None,
         **kwargs,
     ):
+        setup_logzero(level=settings.get("log_level", "info"))
         if minimal_init:
             self.offline_token = settings.get("offline_token")
             self.token_request_url = settings.get("url").get("token_request")

manifester/settings.py

@@ -14,14 +14,14 @@
 
 settings_path = MANIFESTER_DIRECTORY.joinpath("manifester_settings.yaml")
 validators = [
-    # Validator("offline_token", must_exist=True),
+    Validator("offline_token", must_exist=True),
     Validator("simple_content_access", default="enabled"),
     Validator("username_prefix", len_min=3),
 ]
 settings = Dynaconf(
     settings_file=str(settings_path.absolute()),
     ENVVAR_PREFIX_FOR_DYNACONF="MANIFESTER",
+    load_dotenv=True,
     validators=validators,
 )
-
-settings.validators.validate()
+# settings.validators.validate()

manifester_settings.yaml.example

@@ -1,15 +1,18 @@
 #rhsm-manifester settings
+inventory_path: "manifester_inventory.yaml"
 log_level: "info"
 offline_token: ""
 proxies: {"https": ""}
+url:
+  token_request: "https://sso.redhat.com/auth/realms/redhat-external/protocol/openid-connect/token"
+  allocations: "https://api.access.redhat.com/management/v1/allocations"
 username_prefix: "example_username"  # replace value with a unique username
-inventory_path: "manifester_inventory.yaml"
 manifest_category:
   golden_ticket:
     # An offline token can be generated at https://access.redhat.com/management/api
     offline_token: ""
-    # Value of sat_version setting should be in the form 'sat-6.10'
-    sat_version: "sat-6.10"
+    # Value of sat_version setting should be in the form 'sat-6.14'
+    sat_version: "sat-6.14"
     # golden_ticket manifests should not use a quantity higher than 1 for any subscription
     # unless doing so is required for a test.
     subscription_data:
@@ -25,7 +28,7 @@ manifest_category:
     proxies: {"https": ""}
   robottelo_automation:
     offline_token: ""
-    sat_version: "sat-6.10"
+    sat_version: "sat-6.14"
     subscription_data:
       - name: "Software Collections and Developer Toolset"
         quantity: 3

pyproject.toml

@@ -40,7 +40,7 @@ classifiers = [
 ]
 dependencies = [
     "click",
-    "dynaconf",
+    "dynaconf[vault]",
     "logzero",
     "pytest",
     "pyyaml",
@@ -156,6 +156,7 @@ ignore = [
     "D407", # Section name underlining
     "E731", # do not assign a lambda expression, use a def
     "PLR0913", # Too many arguments to function call ({c_args} > {max_args})
+    "PLW1510", # subprocess.run without an explict `check` argument
     "RUF012", # Mutable class attributes should be annotated with typing.ClassVar
     "D107", # Missing docstring in __init__
 ]

scripts/vault_login.py

@@ -0,0 +1,14 @@
+#!/usr/bin/env python
+"""Enables and Disables an OIDC token to access secrets from HashiCorp Vault."""
+import sys
+
+from manifester.helpers import Vault
+
+if __name__ == "__main__":
+    with Vault() as vclient:
+        if sys.argv[-1] == "--login":
+            vclient.login()
+        elif sys.argv[-1] == "--status":
+            vclient.status()
+        else:
+            vclient.logout()