Improved opta ui module (#223)

* Improved opta ui module

1. The service must inherit from the environment always. (sort of-- there is nothing stopping someone from redundantly adding the runx module in the service, but it changes nothing)
2. For the environment, they should only ever need to save the api key once. It should not be required as an envar everytime (stored in AWS parameter store or google secret manager)
3. The service must be able to fetch the stored secret containing the api key
4. It is ok to treat the runx module special (it is, only 6 lines where we run its process or posthook step in the service runs as well)

* linting

* try

* try

Co-authored-by: Nitin Aggarwal <nitin@runx.dev>
Co-authored-by: Nitin Aggarwal <nitin.agg1909@gmail.com>

Author: juandiegopalomino

Pipfile

@@ -34,6 +34,7 @@ requests = ">=2.0,<3"
 kubernetes = ">=12.0,<13"
 python-hcl2 = ">=2.0,<3"
 yamale = ">=3.0,<4"
+google-cloud-secret-manager = "2.4.0"
 google-api-python-client = ">=2.0.2,<2.1.0"
 oauth2client = ">=4.1.3,<4.2.0"
 google-cloud-storage = ">=1.36.2,<1.37.0"

Pipfile.lock

@@ -5,7 +5,6 @@ environments:
       max_nodes: 2
 name: service-1
 modules:
-  - type: runx
   - name: app
     type: k8s-service
     image: kennethreitz/httpbin 

examples/aws/service/opta.yml

@@ -5,7 +5,6 @@ environments:
       max_nodes: 2
 name: service-1
 modules:
-  - type: runx
   - name: app
     type: gcp-k8s-service
     image: kennethreitz/httpbin

examples/gcp/service/opta.yml

@@ -1,6 +1,6 @@
 [mypy]
 warn_redundant_casts = True
-warn_unused_ignores = True
+warn_unused_ignores = False
 warn_unreachable = True
 disallow_untyped_calls = True
 disallow_untyped_defs = True
@@ -10,7 +10,7 @@ disallow_untyped_decorators = False
 no_implicit_optional = True
 strict_optional = True
 strict_equality = True
-implicit_reexport = False
+implicit_reexport = True
 show_error_context = True
 pretty = True
 show_traceback = True

mypy.ini

@@ -3,7 +3,7 @@
 import boto3
 import click
 import yaml
-from google.cloud import storage
+from google.cloud import storage  # type: ignore
 from google.cloud.exceptions import NotFound
 
 from opta.amplitude import amplitude_client

opta/commands/destroy.py

@@ -5,7 +5,7 @@
 from google.auth import default
 from google.auth.credentials import Credentials
 from google.auth.exceptions import DefaultCredentialsError, GoogleAuthError
-from google.cloud import storage
+from google.cloud import storage  # type: ignore
 from google.cloud.exceptions import NotFound
 from google.oauth2 import service_account
 from googleapiclient import discovery

opta/core/gcp.py

@@ -9,7 +9,7 @@
 from botocore.exceptions import ClientError
 from google.api_core.exceptions import ClientError as GoogleClientError
 from google.api_core.exceptions import Conflict
-from google.cloud import storage
+from google.cloud import storage  # type: ignore
 from google.cloud.exceptions import NotFound
 from googleapiclient import discovery
 from googleapiclient.errors import HttpError

opta/core/terraform.py

@@ -222,6 +222,10 @@ def gen_tf(self, module_idx: int) -> Dict[Any, Any]:
                 RunxProcessor(module, self).process(module_idx)
             else:
                 ModuleProcessor(module, self).process(module_idx)
+        if self.parent is not None and self.parent.get_module("runx") is not None:
+            RunxProcessor(self.parent.get_module("runx"), self).process(  # type:ignore
+                module_idx
+            )
         previous_module_reference = None
         for module in self.modules[0 : module_idx + 1]:
             ret = deep_merge(module.gen_tf(depends_on=previous_module_reference), ret)
@@ -251,6 +255,10 @@ def post_hook(self, module_idx: int, exception: Optional[Exception]) -> None:
                 RunxProcessor(module, self).post_hook(module_idx, exception)
             else:
                 ModuleProcessor(module, self).post_hook(module_idx, exception)
+        if self.parent is not None and self.parent.get_module("runx") is not None:
+            RunxProcessor(self.parent.get_module("runx"), self).post_hook(  # type:ignore
+                module_idx, exception
+            )
 
     def metadata_hydration(self) -> Dict[Any, Any]:
         parent_name = self.parent.name if self.parent is not None else "nil"

opta/layer.py

@@ -4,11 +4,18 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Optional, Tuple
 
+import boto3
+import click
 import requests
+from botocore.config import Config
 from getmac import get_mac_address
 from git.config import GitConfigParser
+from google.api_core.exceptions import NotFound
+from google.cloud import secretmanager
+from mypy_boto3_ssm.client import SSMClient
 
 from opta.constants import VERSION
+from opta.core.gcp import GCP
 from opta.exceptions import UserErrors
 from opta.module_processors.base import ModuleProcessor
 from opta.utils import logger
@@ -27,9 +34,6 @@ class RunxProcessor(ModuleProcessor):
     def __init__(self, module: "Module", layer: "Layer"):
         if module.data["type"] != "runx":
             raise Exception(f"The module {module.name} was expected to be of type runx")
-        if os.environ.get("OPTA_API_KEY") is None:
-            raise UserErrors("Need opta api key present for this to run")
-        self.api_key = os.environ.get("OPTA_API_KEY")
         self.user_id = GitConfigParser().get_value("user", "email", "no_user")
         self.device_id = get_mac_address()
         self.os_name = os.name
@@ -38,10 +42,96 @@ def __init__(self, module: "Module", layer: "Layer"):
         super(RunxProcessor, self).__init__(module, layer)
 
     def process(self, module_idx: int) -> None:
-        self.fetch_jwt()
+        logger.info("Checking for runx api key secret")
+        if self.fetch_secret() is None:
+            self.set_secret()
+
+    def fetch_secret(self) -> Optional[str]:
+        if self.layer.cloud == "aws":
+            return self._fetch_aws_secret()
+        elif self.layer.cloud == "google":
+            return self._fetch_gcp_secret()
+        else:
+            raise Exception("Can not handle secrets of type")
+
+    def _fetch_aws_secret(self) -> Optional[str]:
+        providers = self.layer.gen_providers(0)
+        region = providers["provider"]["aws"]["region"]
+        ssm_client: SSMClient = boto3.client("ssm", config=Config(region_name=region))
+        try:
+            parameter = ssm_client.get_parameter(
+                Name=f"/opta-{self.layer.get_env()}/runx-api-key", WithDecryption=True
+            )
+            return parameter["Parameter"]["Value"]
+        except ssm_client.exceptions.ParameterNotFound:
+            return None
+
+    def _fetch_gcp_secret(self) -> Optional[str]:
+        credentials, project_id = GCP.get_credentials()
+        sm_client = secretmanager.SecretManagerServiceClient(credentials=credentials)
+        name = f"projects/{project_id}/secrets/opta-{self.layer.get_env()}-runx-api-key/versions/1"
+        try:
+            # Access the secret version.
+            response = sm_client.access_secret_version(
+                request=secretmanager.AccessSecretVersionRequest({"name": name})
+            )
+            return response.payload.data.decode("UTF-8")
+        except NotFound:
+            return None
+
+    def set_secret(self) -> None:
+        while True:
+            value = click.prompt("Please enter your runx api key", type=str,)
+            try:
+                self.fetch_jwt(value)
+            except UserErrors:
+                logger.warn(
+                    "The api key which you passed was invalid, please provide a valid api key from runx"
+                )
+            else:
+                break
+        if self.layer.cloud == "aws":
+            return self._set_aws_secret(value)
+        elif self.layer.cloud == "google":
+            return self._set_gcp_secret(value)
+        else:
+            raise Exception("Can not handle secrets of type")
+
+    def _set_aws_secret(self, secret: str) -> None:
+        providers = self.layer.gen_providers(0)
+        region = providers["provider"]["aws"]["region"]
+        ssm_client: SSMClient = boto3.client("ssm", config=Config(region_name=region))
+        ssm_client.put_parameter(
+            Name=f"/opta-{self.layer.get_env()}/runx-api-key",
+            Value=secret,
+            Type="SecureString",
+        )
+
+    def _set_gcp_secret(self, secret: str) -> None:
+        credentials, project_id = GCP.get_credentials()
+        sm_client = secretmanager.SecretManagerServiceClient(credentials=credentials)
+        sm_secret = sm_client.create_secret(
+            request=secretmanager.CreateSecretRequest(
+                {
+                    "parent": f"projects/{project_id}",
+                    "secret_id": f"opta-{self.layer.get_env()}-runx-api-key",
+                    "secret": {"replication": {"automatic": {}}},
+                }
+            )
+        )
+        sm_client.add_secret_version(
+            request=secretmanager.AddSecretVersionRequest(
+                {"parent": sm_secret.name, "payload": {"data": secret.encode("utf-8")}}
+            )
+        )
 
     def post_hook(self, module_idx: int, exception: Optional[Exception]) -> None:
-        validation_data, jwt = self.fetch_jwt()
+        api_key = self.fetch_secret()
+        if api_key is None:
+            raise Exception(
+                "The api key seems to have just disappeared from the secret storage"
+            )
+        validation_data, jwt = self.fetch_jwt(api_key)
         is_environment = self.layer.parent is None
         url_path = "/config/environments" if is_environment else "/config/services"
         body = {
@@ -73,9 +163,9 @@ def post_hook(self, module_idx: int, exception: Optional[Exception]) -> None:
                 f"Invalid response when attempting to send data to backend: {resp.json()}"
             )
 
-    def fetch_jwt(self) -> Tuple[dict, str]:
+    def fetch_jwt(self, api_key: str) -> Tuple[dict, str]:
         resp = requests.post(
-            f"https://{OPTA_DOMAIN}/user/apikeys/validate", json={"api_key": self.api_key}
+            f"https://{OPTA_DOMAIN}/user/apikeys/validate", json={"api_key": api_key}
         )
         if resp.status_code == 404:
             raise UserErrors(