gui: add methods for reloading config files and clean up secret manag…

…ers.

unfurl/localenv.py

@@ -117,6 +117,15 @@ def _set_parent_project(
         # depends on _set_contexts():
         self.project_repoview.yaml = make_yaml(self.make_vault_lib())
 
+    def reload(self):
+        config = self.localConfig.config
+        self.localConfig = LocalConfig(
+            config.path, yaml_include_hook=config.loadHook, readonly=config.readonly
+        )
+        self._set_contexts()
+        # depends on _set_contexts():
+        self.project_repoview.yaml = make_yaml(self.make_vault_lib())
+
     def _set_project_repoview(self) -> None:
         path = self.projectRoot
         if path in self.workingDirs:

unfurl/server/gui.py

@@ -310,14 +310,13 @@ def get_variables(project_path):
 
     @app.route("/<path:project_path>/-/variables", methods=["PATCH"])
     def patch_variables(project_path):
-        nonlocal localenv
         repo = get_repo(project_path)
         if not repo or repo.repo != localrepo.repo:
             return notfound_response(project_path)
 
         body = request.json
         if isinstance(body, dict) and "variables_attributes" in body:
-            localenv = set_variables(localenv, body["variables_attributes"])
+            set_variables(localenv, body["variables_attributes"])
             return {"variables": list(yield_variables(localenv))}
         else:
             return "Bad Request", 400

unfurl/server/gui_variables/__init__.py

@@ -1,35 +1,40 @@
 import re
 from functools import lru_cache
-from typing import Iterator, List
+from typing import Any, Iterator, List, Literal, Union
+from typing_extensions import TypedDict, Required
 
-from .envvar import EnvVar
 from ...localenv import LocalEnv
 from ..serve import app
 
 from . import local_secrets
 from . import ufcloud_secrets
 
 
-@lru_cache
-def _get_secrets_manager(localenv):
-    url, _ = localenv.project.localConfig.config.search_includes(
-        pathPrefix=app.config["UNFURL_CLOUD_SERVER"]
-    )
+class EnvVar(TypedDict, total=False):
+    # see https://docs.gitlab.com/ee/api/project_level_variables.html
+    id: Union[int, str]  # ID or URL-encoded path of the project
+    key: Required[str]
+    masked: Required[bool]
+    environment_scope: Required[str]
+    value: Any
+    secret_value: Any  # ??? not sent by api
+    _destroy: bool
+    variable_type: Required[Union[Literal["env_var"], Literal["file"]]]
+    raw: Literal[False]  # if true value isn't expanded
+    protected: Literal[False]
 
-    gitlab_api_match = re.search(
-        r"/api/v4/projects/(?P<project_id>(\w|%2F|[/\-_])+)/variables\?[^&]*&private_token=(?P<private_token>[^&]+)",
-        str(url),
-    )
 
-    if gitlab_api_match:
+@lru_cache
+def _get_secrets_manager(localenv: LocalEnv):
+    url, project_id = ufcloud_secrets.find_gitlab_endpoint(localenv)
+    if project_id:
         return ufcloud_secrets
-    else:
-        return local_secrets
+    return local_secrets
 
 
-def set_variables(localenv, env_vars: List[EnvVar]) -> LocalEnv:
+def set_variables(localenv: LocalEnv, env_vars: List[EnvVar]):
     return _get_secrets_manager(localenv).set_variables(localenv, env_vars)
 
 
-def yield_variables(localenv) -> Iterator[EnvVar]:
+def yield_variables(localenv: LocalEnv) -> Iterator[EnvVar]:
     return _get_secrets_manager(localenv).yield_variables(localenv)

unfurl/server/gui_variables/local_secrets.py

@@ -1,5 +1,5 @@
 import os
-from typing import Iterator, List, Literal, Union
+from typing import Any, Dict, Iterator, List, Literal, Optional, Union, cast
 
 from ...logs import is_sensitive, getLogger
 
@@ -14,68 +14,84 @@
 logger = getLogger("unfurl.gui")
 
 
-def set_variables(localenv, env_vars: List[EnvVar]) -> LocalEnv:
+def _get_env_vars(envs: Dict[str, dict], env_name: str) -> Dict[str, Any]:
+    env = envs.get(env_name)
+    if env:
+        return env.get("variables", {})
+    return {}
+
+
+def _set_env_var(environments: Dict[str, dict], env_name: str, key: str, val: Any) -> Dict[str, Any]:
+    env = environments.setdefault(env_name, CommentedMap())
+    variables = env.setdefault("variables", CommentedMap())
+    variables[key] = val
+    return variables
+
+
+def set_variables(localenv: LocalEnv, env_vars: List[EnvVar]) -> None:
     # reload immediately in case of user edits
-    localenv = LocalEnv(UNFURL_SERVE_PATH, overrides={"ENVIRONMENT": "*"})
     project = localenv.project or localenv.homeProject
     assert project
-    secret_config_key, secret_config = project.localConfig.find_secret_include()
-    secret_environments = (
-        secret_config.setdefault("environments", CommentedMap())
-        if secret_config
-        else None
-    )
-    config: dict = project.localConfig.config.config
+    project.reload()
+
+    config = cast(Dict[str, Dict], project.localConfig.config.config)
     assert config
-    env = config.setdefault("environments", CommentedMap())
+    envs = config.setdefault("environments", CommentedMap())
+
+    secret_config_key, secret_config = project.localConfig.find_secret_include()
+    if secret_config is not None:
+        secret_environments = cast(
+            Optional[Dict[str, Dict]],
+            secret_config.setdefault("environments", CommentedMap()),
+        )
+    else:
+        secret_environments = None
+
     modified_secrets = False
     modified_config = False
     for envvar in env_vars:
         environment_scope = envvar["environment_scope"]
         env_name = "defaults" if environment_scope == "*" else environment_scope
+        config_env_vars = _get_env_vars(envs, env_name)
+        if secret_environments is not None:
+            secret_env_vars = _get_env_vars(secret_environments, env_name)
+        else:
+            secret_env_vars = {}
         key = envvar["key"]
         value = envvar.get("secret_value", envvar.get("value"))
         if envvar["variable_type"] == "file":
             value = {"eval": dict(tempfile=value)}
         if envvar.get("_destroy"):
-            if env_name in env and key in env[env_name]:
+            if key in config_env_vars:
                 modified_config = True
-                del env[env_name][key]
-            secret_env = (
-                secret_environments.get(env_name) if secret_environments else None
-            )
-            if secret_env and key in secret_env.get("variables", {}):
+                del config_env_vars[key]
+            if key in secret_env_vars:
                 modified_secrets = True
-                del secret_env["variables"][key]
+                del secret_env_vars[key]
         else:
             if envvar["masked"]:
-                env.pop(key, None)  # in case this flag changed
-                if secret_environments is not None:
-                    secret_env = secret_environments.setdefault(
-                        env_name, CommentedMap()
-                    )
+                secret_val = {"eval": dict(sensitive=value)}  # mark sensitive
+                if secret_environments:
+                    _set_env_var(secret_environments, env_name, key, secret_val)
                     modified_secrets = True
-                    secret_env.setdefault("variables", {})[key] = {
-                        "eval": dict(sensitive=value)
-                    }  # mark sensitive
+                    if key in config_env_vars:
+                        modified_config = True
+                        del config_env_vars[key]
+                else:
+                    _set_env_var(envs, env_name, key, secret_val)
+                    modified_config = True
             else:
-                # in case this flag changed
-                secret_env = (
-                    secret_environments.get(env_name) if secret_environments else None
-                )
-                if secret_env and key in secret_env.get("variables", {}):
+                if key in secret_env_vars:  # in case masked flag changed
                     modified_secrets = True
-                    del secret_env["variables"][key]
+                    del secret_env_vars[key]
+                _set_env_var(envs, env_name, key, value)
                 modified_config = True
-                env.setdefault(env_name, CommentedMap())[key] = value
     if modified_secrets:
         project.localConfig.config.save_include(secret_config_key)
     if modified_config:
         project.localConfig.config.save()
     if modified_secrets or modified_config:
-        # reload
-        localenv = LocalEnv(UNFURL_SERVE_PATH, overrides={"ENVIRONMENT": "*"})
-    return localenv
+        project.reload()
 
 
 def yield_variables(localenv) -> Iterator[EnvVar]:

unfurl/server/gui_variables/ufcloud_secrets.py

@@ -1,5 +1,5 @@
 import os
-from typing import Iterator, List
+from typing import Iterator, List, Optional, Tuple
 import re
 from . import EnvVar
 from ..serve import app
@@ -8,40 +8,53 @@
 import gitlab
 from functools import lru_cache
 
-UNFURL_SERVE_PATH = os.getenv("UNFURL_SERVE_PATH", "")
 
-
-@lru_cache
-def _get_context(localenv):
+def find_gitlab_endpoint(localenv: LocalEnv) -> Tuple[Optional[str], Optional[str]]:
+    if not localenv.project:
+        return None, None
     url, _ = localenv.project.localConfig.config.search_includes(
         pathPrefix=app.config["UNFURL_CLOUD_SERVER"]
     )
-
-    parsed_url = urllib.parse.urlparse(url)
-    query_params = urllib.parse.parse_qs(parsed_url.query)
-    private_token = query_params.get("private_token", [None])[0]
+    if not url:
+        return url, None
     project_id_match = re.search(
-        r"projects/(?P<project_id>(\w|%2F|[/\-_])+)/variables", parsed_url.path
+        r"/api/v4/projects/(?P<project_id>(\w|%2F|[/\-_])+)/variables", url
     )
+    if not project_id_match:
+        return url, None
+    project_id = project_id_match.group("project_id").replace("%2F", "/")
+    return url, project_id
 
-    if project_id_match and private_token:
-        project_id = project_id_match.group("project_id").replace("%2F", "/")
-        origin = f"{parsed_url.scheme}://{parsed_url.hostname}"
 
-        gl = gitlab.Gitlab(origin, private_token=private_token)
-        gl.auth()
-        gl.enable_debug()
-        project = gl.projects.get(project_id)
+@lru_cache
+def _get_context(localenv: LocalEnv):
+    url, project_id = find_gitlab_endpoint(localenv)
+    if not url or not project_id:
+        return None
 
-        return (gl, project)
-    else:
-        raise ValueError(
-            f"Could not access project_id and/or private_token from url '{url}'"
-        )
+    parsed_url = urllib.parse.urlparse(url)
+    if not parsed_url.query:
+        return None
+
+    query_params = urllib.parse.parse_qs(str(parsed_url.query))
+    private_token = query_params.get("private_token", [""])[0]
+    if not private_token:
+        return None
+
+    origin = f"{parsed_url.scheme}://{parsed_url.hostname}"
 
+    gl = gitlab.Gitlab(origin, private_token=private_token)
+    gl.auth()
+    gl.enable_debug()
+    project = gl.projects.get(project_id)
 
-def set_variables(localenv, env_vars: List[EnvVar]) -> LocalEnv:
-    _, project = _get_context(localenv)
+    return project
+
+
+def set_variables(localenv: LocalEnv, env_vars: List[EnvVar]) -> None:
+    project = _get_context(localenv)
+    if not project:
+        raise ValueError("Could not access project_id and/or private_token from url")
 
     for var in env_vars:
         data = {
@@ -58,11 +71,11 @@ def set_variables(localenv, env_vars: List[EnvVar]) -> LocalEnv:
         else:
             project.variables.create(data)
 
-    return localenv
-
 
-def yield_variables(localenv) -> Iterator[EnvVar]:
-    _, project = _get_context(localenv)
+def yield_variables(localenv: LocalEnv) -> Iterator[EnvVar]:
+    project = _get_context(localenv)
+    if not project:
+        raise ValueError("Could not access project_id and/or private_token from url")
 
     for variable in project.variables.list(get_all=True):
         yield EnvVar(

unfurl/yamlloader.py

@@ -681,7 +681,9 @@ def resolve_url(
                 url = base
             else:
                 # url is a local path
-                assert base or os.path.isabs(file_name), f"{file_name} isn't absolute and base isn't set"
+                assert base or os.path.isabs(
+                    file_name
+                ), f"{file_name} isn't absolute and base isn't set"
                 url = os.path.join(base, file_name)
                 repository_root = None  # default to checking if its in the project
                 if importsLoader.repository_root:
@@ -1031,7 +1033,7 @@ def __init__(
 
             # schema should include defaults but can't validate because it doesn't understand includes
             # but should work most of time
-            self.config.loadTemplate = self.load_include  # type: ignore
+            setattr(self.config, "loadTemplate", self.load_include)
             self.loadHook = loadHook
             self.baseDirs = [self.get_base_dir()]
             while True:
@@ -1056,6 +1058,18 @@ def __init__(
         except Exception:
             raise UnfurlBadDocumentError(err_msg, saveStack=True)
 
+    def clone(self, validate: bool = True) -> "YamlConfig":
+        # reloads the config
+        return YamlConfig(
+            self.config,
+            self.path,
+            validate,
+            self.schema,
+            self.loadHook,
+            self.vault,
+            self.readonly,
+        )
+
     def _expand(self) -> Tuple[Mapping, Mapping]:
         find_anchor(self.config, None)  # create _anchorCache
         self._cachedDocIncludes: Dict[str, Tuple[str, dict]] = {}
@@ -1160,7 +1174,9 @@ def validate(self, config):
             baseUri = urljoin("file:", urllib.request.pathname2url(path))
         return find_schema_errors(config, self.schema, baseUri)
 
-    def search_includes(self, key: Optional[str]=None, pathPrefix: Optional[str]=None) -> Union[Tuple[None, None], Tuple[str, dict]]:
+    def search_includes(
+        self, key: Optional[str] = None, pathPrefix: Optional[str] = None
+    ) -> Union[Tuple[None, None], Tuple[str, dict]]:
         for k in self._cachedDocIncludes:
             path, template = self._cachedDocIncludes[k]
             candidate = True
@@ -1255,7 +1271,7 @@ def load_include(
             msg = f"unable to load document include: {templatePath} (base: {baseDir})"
             if warnWhenNotFound:
                 logger.warning(msg, exc_info=True)
-                template = None
+                return value, None, baseDir
             else:
                 raise UnfurlError(
                     msg,