Add the ability to specify custom build steps when adding modules

cfbs/cfbs_config.py

@@ -19,7 +19,7 @@
 import glob
 import logging as log
 from collections import OrderedDict
-from typing import List
+from typing import List, Optional
 
 from cfbs.types import CFBSCommandGitResult
 from cfbs.utils import (
@@ -37,7 +37,7 @@
 )
 from cfbs.pretty import pretty, CFBS_DEFAULT_SORTING_RULES
 from cfbs.cfbs_json import CFBSJson
-from cfbs.module import Module, is_module_added_manually
+from cfbs.module import Module, is_module_added_manually, is_module_local
 from cfbs.prompts import prompt_user, prompt_user_yesno
 from cfbs.validate import validate_single_module
 
@@ -106,17 +106,15 @@ def longest_module_key_length(self, key) -> int:
             else 0
         )
 
-    def add_with_dependencies(
-        self, module, remote_config=None, str_added_by="cfbs add"
-    ):
+    def add_with_dependencies(self, module, remote_config=None, added_by="cfbs add"):
         if type(module) is list:
             # TODO: reuse logic from _add_modules instead
             for m in module:
-                self.add_with_dependencies(m, remote_config, str_added_by)
+                self.add_with_dependencies(m, remote_config, added_by)
             return
         if type(module) is str:
             module_str = module
-            module = (remote_config or self).get_module_for_build(module, str_added_by)
+            module = (remote_config or self).get_module_for_build(module, added_by)
             if not module:
                 raise CFBSExitError("Module '%s' not found" % module_str)
         if not module:
@@ -125,7 +123,7 @@ def add_with_dependencies(
         name = module["name"]
         assert "steps" in module
         if self._module_is_in_build(module):
-            if is_module_added_manually(str_added_by):
+            if is_module_added_manually(added_by):
                 print("Skipping already added module '%s'" % name)
             return
         if "dependencies" in module:
@@ -145,10 +143,11 @@ def add_with_dependencies(
 
     def _add_using_url(
         self,
-        url,
+        url: str,
         to_add: list,
         added_by="cfbs add",
         checksum=None,
+        explicit_build_steps=None,
     ):
         url_commit = None
         if url.endswith(SUPPORTED_ARCHIVES):
@@ -195,7 +194,7 @@ def _add_using_url(
                     % (i, len(modules), module["name"]),
                 ):
                     continue
-            self.add_with_dependencies(module, remote_config, str_added_by=added_by)
+            self.add_with_dependencies(module, remote_config, added_by)
 
     @staticmethod
     def _convert_added_by(added_by, to_add):
@@ -297,7 +296,7 @@ def _add_bundles_build_step(self, module, policy_files):
         module["steps"].append(step)
         log.debug("Added build step '%s' for module '%s'" % (step, name))
 
-    def _handle_local_module(self, module):
+    def _handle_local_module(self, module, use_default_build_steps=True):
         name = module["name"]
         if not (
             name.startswith("./")
@@ -329,10 +328,17 @@ def _handle_local_module(self, module):
                     )
                 # TODO: Support adding local modules with autorun tag
 
-        self._add_policy_files_build_step(module)
-        self._add_bundles_build_step(module, policy_files)
+        if use_default_build_steps:
+            self._add_policy_files_build_step(module)
+            self._add_bundles_build_step(module, policy_files)
+
+    def _add_without_dependencies(self, modules, use_default_build_steps=True):
+        """Note: `use_default_build_steps` is only relevant for local modules."""
+        if not use_default_build_steps:
+            assert len(modules) == 1 and modules[0]["name"].startswith(
+                "./"
+            ), "`use_default_build_steps` is currently only expected to be explicitly used for adding a single local module"
 
-    def _add_without_dependencies(self, modules):
         assert modules
         assert len(modules) > 0
         assert modules[0]["name"]
@@ -350,7 +356,7 @@ def _add_without_dependencies(self, modules):
                 context="build", name=name, module=module, config=None, local_check=True
             )
             self["build"].append(module)
-            self._handle_local_module(module)
+            self._handle_local_module(module, use_default_build_steps)
 
             assert "added_by" in module
             added_by = module["added_by"]
@@ -364,7 +370,17 @@ def _add_modules(
         to_add: list,
         added_by="cfbs add",
         checksum=None,
-    ) -> int:
+        explicit_build_steps: Optional[List[str]] = None,
+    ):
+        """Note: explicit build steps are currently limited to single local modules without dependencies, see the asserts."""
+        if explicit_build_steps is not None:
+            assert (
+                len(to_add) == 1
+            ), "explicit_build_steps is only for adding a single module"
+            assert is_module_local(
+                to_add[0]
+            ), "explicit_build_steps is only for adding a local module"
+
         index = self.index
 
         modules = [Module(m) for m in to_add]
@@ -380,10 +396,14 @@ def _add_modules(
         # Filter modules which are already added:
         to_add = self._filter_modules_to_add(modules)
         if not to_add:
-            return 0  # Everything already added
+            # Everything already added
+            return
 
         # Convert names to objects:
-        modules_to_add = [index.get_module_object(m, added_by[m.name]) for m in to_add]
+        modules_to_add = [
+            index.get_module_object(m, added_by[m.name], explicit_build_steps)
+            for m in to_add
+        ]
         modules_already_added = self["build"]
 
         assert not any(m for m in modules_to_add if "name" not in m)
@@ -393,16 +413,21 @@ def _add_modules(
         dependencies = self._find_dependencies(modules_to_add, modules_already_added)
 
         if dependencies:
+            assert (
+                explicit_build_steps is None
+            ), "explicit build steps do not apply to dependencies"
             self._add_without_dependencies(dependencies)
 
-        self._add_without_dependencies(modules_to_add)
-        return 0
+        self._add_without_dependencies(
+            modules_to_add, use_default_build_steps=explicit_build_steps is None
+        )
 
     def add_command(
         self,
         to_add: List[str],
         added_by="cfbs add",
         checksum=None,
+        explicit_build_steps: Optional[List[str]] = None,
     ) -> CFBSCommandGitResult:
         if not to_add:
             raise CFBSUserError("Must specify at least one module to add")
@@ -417,6 +442,7 @@ def add_command(
                 to_add=to_add[1:],
                 added_by=added_by,
                 checksum=checksum,
+                explicit_build_steps=explicit_build_steps,
             )
         else:
             # for this `if` to be valid, module names containing `://` should be illegal
@@ -425,7 +451,7 @@ def add_command(
                     "URI scheme not supported. The supported URI schemes are: "
                     + ", ".join(SUPPORTED_URI_SCHEMES)
                 )
-            self._add_modules(to_add, added_by, checksum)
+            self._add_modules(to_add, added_by, checksum, explicit_build_steps)
 
         added = {m["name"] for m in self["build"]}.difference(before)
 

cfbs/commands.py

@@ -51,7 +51,7 @@ def search_command(terms: List[str]):
 import logging as log
 import json
 import functools
-from typing import Callable, List, Union
+from typing import Callable, List, Optional, Union
 from collections import OrderedDict
 from cfbs.analyze import analyze_policyset
 from cfbs.args import get_args
@@ -399,10 +399,11 @@ def add_command(
     to_add: List[str],
     added_by="cfbs add",
     checksum=None,
+    explicit_build_steps: Optional[List[str]] = None,
 ):
     config = CFBSConfig.get_instance()
     validate_config_raise_exceptions(config, empty_build_list_ok=True)
-    r = config.add_command(to_add, added_by, checksum)
+    r = config.add_command(to_add, added_by, checksum, explicit_build_steps)
     config.save()
     return r
 

cfbs/index.py

@@ -1,7 +1,7 @@
 import sys
 import os
 from collections import OrderedDict
-from typing import Union
+from typing import List, Optional, Union
 
 from cfbs.module import Module
 from cfbs.utils import CFBSNetworkError, get_or_read_json, CFBSExitError, get_json
@@ -15,48 +15,56 @@
 )
 
 
-def _local_module_data_cf_file(module):
-    dst = os.path.join("services", "cfbs", module[2:])
+def _local_module_data_cf_file(module_name: str):
+    dst = os.path.join("services", "cfbs", module_name[2:])
     return {
         "description": "Local policy file added using cfbs command line",
         "tags": ["local"],
-        "steps": ["copy %s %s" % (module, dst)],
+        "steps": ["copy %s %s" % (module_name, dst)],
         "added_by": "cfbs add",
     }
 
 
-def _local_module_data_json_file(module):
+def _local_module_data_json_file(module_name: str):
     return {
         "description": "Local augments file added using cfbs command line",
         "tags": ["local"],
-        "steps": ["json %s def.json" % module],
+        "steps": ["json %s def.json" % module_name],
         "added_by": "cfbs add",
     }
 
 
-def _local_module_data_subdir(module):
-    assert module.startswith("./")
-    assert module.endswith(("/", "/."))
-    dst = os.path.join("services", "cfbs", module[2:])
+def _local_module_data_subdir(
+    module_name: str, explicit_build_steps: Optional[List[str]] = None
+):
+    assert module_name.startswith("./")
+    assert module_name.endswith(("/", "/."))
+    dst = os.path.join("services", "cfbs", module_name[2:])
+    if explicit_build_steps is None:
+        build_steps = ["directory ./ {}".format(dst)]
+    else:
+        build_steps = explicit_build_steps
     return {
         "description": "Local subdirectory added using cfbs command line",
         "tags": ["local"],
-        "steps": ["directory ./ {}".format(dst)],
+        "steps": build_steps,
         "added_by": "cfbs add",
     }
 
 
-def _generate_local_module_object(module):
-    assert module.startswith("./")
-    assert module.endswith((".cf", ".json", "/"))
-    assert os.path.isfile(module) or os.path.isdir(module)
+def _generate_local_module_object(
+    module_name: str, explicit_build_steps: Optional[List[str]] = None
+):
+    assert module_name.startswith("./")
+    assert module_name.endswith((".cf", ".json", "/"))
+    assert os.path.isfile(module_name) or os.path.isdir(module_name)
 
-    if os.path.isdir(module):
-        return _local_module_data_subdir(module)
-    if module.endswith(".cf"):
-        return _local_module_data_cf_file(module)
-    if module.endswith(".json"):
-        return _local_module_data_json_file(module)
+    if os.path.isdir(module_name):
+        return _local_module_data_subdir(module_name, explicit_build_steps)
+    if module_name.endswith(".cf"):
+        return _local_module_data_cf_file(module_name)
+    if module_name.endswith(".json"):
+        return _local_module_data_json_file(module_name)
 
 
 class Index:
@@ -165,15 +173,20 @@ def translate_alias(self, module: Module):
             if os.path.exists(module.name):
                 module.name = local_module_name(module.name)
 
-    def get_module_object(self, module, added_by=None):
+    def get_module_object(
+        self,
+        module,
+        added_by: Optional[str] = None,
+        explicit_build_steps: Optional[List[str]] = None,
+    ):
         if isinstance(module, str):
             module = Module(module)
         name = module.name
         version = module.version
         module = module.to_dict()
 
         if name.startswith("./"):
-            object = _generate_local_module_object(name)
+            object = _generate_local_module_object(name, explicit_build_steps)
         else:
             object = self[name]
             if version:

cfbs/internal_file_management.py

@@ -155,7 +155,7 @@ def _clone_and_checkout(url, path, commit):
     sh("git checkout " + commit, directory=path)
 
 
-def clone_url_repo(repo_url):
+def clone_url_repo(repo_url: str):
     assert repo_url.startswith(SUPPORTED_URI_SCHEMES)
 
     commit = None
@@ -196,7 +196,7 @@ def clone_url_repo(repo_url):
 
 
 def fetch_archive(
-    url, checksum=None, directory=None, with_index=True, extract_to_directory=False
+    url: str, checksum=None, directory=None, with_index=True, extract_to_directory=False
 ):
     assert url.endswith(SUPPORTED_ARCHIVES)