feat: add ubuntu provider git config options (#49)

Adds configuration options to the ubuntu driver to specify the `git_url`
and `git_branch` to use for the ubuntu-cve-tracker repository

Signed-off-by: Weston Steimel <[email protected]>

src/vunnel/providers/ubuntu/__init__.py

@@ -6,7 +6,7 @@
 
 from vunnel import provider, result, schema
 
-from .parser import Parser, default_max_workers
+from .parser import Parser, default_git_branch, default_git_url, default_max_workers
 
 
 @dataclass
@@ -21,6 +21,8 @@ class Config:
     additional_versions: dict[str, str] = field(default_factory=lambda: {})
     enable_rev_history: bool = True
     max_workers: int = default_max_workers
+    git_url: str = default_git_url
+    git_branch: str = default_git_branch
 
 
 class Provider(provider.Provider):
@@ -37,6 +39,8 @@ def __init__(self, root: str, config: Config):
             additional_versions=self.config.additional_versions,
             enable_rev_history=self.config.enable_rev_history,
             max_workers=self.config.max_workers,
+            git_url=self.config.git_url,
+            git_branch=self.config.git_branch,
         )
 
     @classmethod

src/vunnel/providers/ubuntu/git.py

@@ -31,19 +31,22 @@ class GitWrapper:
     __cve_id_regex__ = re.compile(r"CVE-\S+")
     _check_cmd_ = "git --version"
     _is_git_repo_cmd_ = "git rev-parse --is-inside-work-tree"
-    _clone_cmd_ = "git clone -b master {src} {dest}"
-    _check_out_cmd_ = "git checkout master"
+    _set_remote_cmd_ = "git remote set-url origin {src}"
+    _clone_cmd_ = "git clone -b {branch} {src} {dest}"
+    _check_out_cmd_ = "git checkout {branch}"
     _pull_cmd_ = "git pull -f"
     _fetch_cmd_ = "git fetch --all"
     _pull_ff_only_cmd_ = "git pull --ff-only"
-    _reset_head_cmd_ = "git reset --hard origin/master"
     _write_graph_ = "git commit-graph write --reachable --changed-paths"
     _change_set_cmd_ = "git log --no-renames --no-merges --name-status --format=oneline {from_rev}..{to_rev}"
     _get_rev_content_cmd_ = "git show {sha}:{file}"
     _head_rev_cmd_ = "git rev-parse HEAD"
 
-    def __init__(self, source: str, checkout_dest: str, workspace: str | None = None, logger: logging.Logger | None = None):
+    def __init__(
+        self, source: str, branch: str, checkout_dest: str, workspace: str | None = None, logger: logging.Logger | None = None
+    ):
         self.src = source
+        self.branch = branch
         self.dest = checkout_dest
         self.workspace = workspace if workspace else tempfile.gettempdir()
 
@@ -83,15 +86,15 @@ def init_repo(self, force=False):
             return
 
         try:
-            self.logger.info("cloning git repository {} to {}".format(self.src, self.dest))
+            self.logger.info(f"cloning git repository {self.src} branch {self.branch} to {self.dest}")
 
-            cmd = self._clone_cmd_.format(src=self.src, dest=self.dest)
+            cmd = self._clone_cmd_.format(src=self.src, dest=self.dest, branch=self.branch)
             out = self._exec_cmd(cmd)
 
             self.logger.debug("initialized git repo, cmd: {}, output: {}".format(cmd, out.decode()))
             self._write_graph()
         except:
-            self.logger.exception("failed to clone initialize git repository {} to {}".format(self.src, self.dest))
+            self.logger.exception(f"failed to clone git repository {self.src} branch {self.branch} to {self.dest}")
             raise
 
     def parse_full_cve_revision_history(self, git_log_output: str) -> dict[str, list[GitRevision]]:
@@ -117,7 +120,8 @@ def prepare_cve_revision_history(self):
     def sync_with_upstream(self):
         try:
             try:
-                self._exec_cmd(self._check_out_cmd_, cwd=self.dest)
+                self._exec_cmd(self._set_remote_cmd_.format(src=self.src), cwd=self.dest)
+                self._exec_cmd(self._check_out_cmd_.format(branch=self.branch), cwd=self.dest)
             except:  # nosec
                 pass
             out = self._exec_cmd(self._pull_ff_only_cmd_, cwd=self.dest)

src/vunnel/providers/ubuntu/parser.py

@@ -21,6 +21,8 @@
 namespace = "ubuntu"
 
 default_max_workers = 8
+default_git_url = "git://git.launchpad.net/ubuntu-cve-tracker"
+default_git_branch = "master"
 
 ubuntu_pkg_version_format = "dpkg"
 ubuntu_cve_url = "http://people.ubuntu.com/~ubuntu-security/cve/{}"
@@ -580,8 +582,7 @@ class Parser:
     __payload__ = Vulnerability
 
     _bzr_src = "https://launchpad.net/ubuntu-cve-tracker"
-    _git_https = "https://git.launchpad.net/ubuntu-cve-tracker"
-    _git_src = "git://git.launchpad.net/ubuntu-cve-tracker"
+    _git_src_url = "git://git.launchpad.net/ubuntu-cve-tracker"
     _bzr_to_git_transition_commit = "dc3f64a0dfe6b1780240ff115d8a0a1b23fd00b4"
 
     _active_cve_dir = "active"
@@ -605,16 +606,19 @@ def __init__(
         additional_versions: dict[str, str] | None = None,
         enable_rev_history: bool = True,
         max_workers: int = default_max_workers,
+        git_url: str = default_git_url,
+        git_branch: str = default_git_branch,
     ):
         self.vc_workspace = os.path.join(workspace.input_path, self._vc_working_dir)
         # TODO: tech debt: this should use the results workspace with the correct schema-aware envelope
         self.norm_workspace = os.path.join(workspace.input_path, self._normalized_cve_dir)
         if not logger:
             logger = logging.getLogger(self.__class__.__name__)
         self.logger = logger
-        self.urls = [self._git_https]
-
-        self.git_wrapper = GitWrapper(source=self._git_src, checkout_dest=self.vc_workspace, logger=logger)
+        self.git_url = git_url
+        self.git_branch = git_branch
+        self.urls = [self.git_url]
+        self.git_wrapper = GitWrapper(source=self.git_url, branch=self.git_branch, checkout_dest=self.vc_workspace, logger=logger)
 
         if additional_versions:
             ubuntu_version_names.update(additional_versions)

tests/unit/cli/test-fixtures/full.yaml

@@ -52,6 +52,15 @@ providers:
     request_timeout: 20
     allow_versions:
       - 13
+  ubuntu:
+    runtime: *runtime
+    request_timeout: 20
+    additional_versions:
+      "zzz": "24.24"
+    enable_rev_history: true
+    max_workers: 25
+    git_url: "https://xyz.abc"
+    git_branch: "yoda"
   wolfi:
     runtime: *runtime
     request_timeout: 20

tests/unit/cli/test_cli.py

@@ -257,6 +257,8 @@ def test_config(monkeypatch) -> None:
   ubuntu:
     additional_versions: {}
     enable_rev_history: true
+    git_branch: master
+    git_url: git://git.launchpad.net/ubuntu-cve-tracker
     max_workers: 8
     request_timeout: 125
     runtime:

tests/unit/cli/test_config.py

@@ -79,6 +79,15 @@ def test_full_config(helpers):
                 request_timeout=20,
                 allow_versions=[13],
             ),
+            ubuntu=providers.ubuntu.Config(
+                runtime=runtime_cfg,
+                request_timeout=20,
+                additional_versions={"zzz": "24.24"},
+                enable_rev_history=True,
+                max_workers=25,
+                git_url="https://xyz.abc",
+                git_branch="yoda",
+            ),
             wolfi=providers.wolfi.Config(
                 runtime=runtime_cfg,
                 request_timeout=20,

tests/unit/providers/ubuntu/test_git_wrapper.py

@@ -133,7 +133,7 @@ def test_parse_log(self):
         with open(self._git_change_log_file_) as f:
             git_commit_log = f.read()
 
-        wrapper = GitWrapper(self._workspace_, self._workspace_)
+        wrapper = GitWrapper(self._workspace_, "master", self._workspace_)
 
         commits = wrapper._parse_log(git_commit_log)
 
@@ -149,7 +149,7 @@ def test_compute_change_set(self):
         with open(self._git_change_log_file_) as f:
             git_commit_log = f.read()
 
-        wrapper = GitWrapper(self._workspace_, self._workspace_)
+        wrapper = GitWrapper(self._workspace_, "master", self._workspace_)
 
         commits = wrapper._parse_log(git_commit_log)
 
@@ -195,4 +195,4 @@ def test_compute_change_set(self):
     ],
 )
 def test_parse_full_cve_revision_history(git_log_output: str, expected: dict[str, list[GitRevision]]):
-    assert GitWrapper("", "").parse_full_cve_revision_history(git_log_output) == expected
+    assert GitWrapper("", "master", "").parse_full_cve_revision_history(git_log_output) == expected