fix: harden ini config validation and overrides

Agent-Logs-Url: https://github.com/VulnerabilityHistoryProject/recidivism/sessions/a059b408-391c-467d-b495-6d9d3789015b

Co-authored-by: andymeneely <341847+andymeneely@users.noreply.github.com>

Author: Copilot

scripts/clone_osv_repositories.py

@@ -5,7 +5,7 @@
 from urllib.parse import urlparse
 
 from osv_common import extract_repo_urls, iter_vulnerability_files, load_vulnerability
-from recidivism_config import load_config_with_source, required_value, resolve_config_path
+from recidivism_config import get_required_value, load_config_with_source, resolve_config_path
 
 
 def clone_or_update(repo_url: str, target_dir: Path, update_existing: bool) -> None:
@@ -46,13 +46,18 @@ def main() -> None:
     config, config_source = load_config_with_source("clone")
 
     parser = argparse.ArgumentParser(description="Clone all repositories referenced by OSV vulnerabilities.")
-    parser.add_argument("--osv-dir", help="Directory containing extracted OSV JSON files")
-    parser.add_argument("--target-dir", help="Directory to place local repository clones")
-    max_repos_str = config.get("max_repos", fallback="").strip()
+    parser.add_argument(
+        "--osv-dir",
+        help="Directory containing extracted OSV JSON files (overrides clone.osv_dir in recidivism.ini)",
+    )
+    parser.add_argument(
+        "--target-dir",
+        help="Directory to place local repository clones (overrides clone.target_dir in recidivism.ini)",
+    )
     parser.add_argument(
         "--max-repos",
         type=int,
-        default=int(max_repos_str) if max_repos_str else None,
+        default=None,
         help="Optional limit for number of repositories",
     )
     parser.add_argument(
@@ -64,10 +69,18 @@ def main() -> None:
     args = parser.parse_args()
 
     try:
-        osv_dir = resolve_config_path(args.osv_dir or required_value(config, "osv_dir"))
-        target_dir = resolve_config_path(args.target_dir or required_value(config, "target_dir"))
+        osv_dir = resolve_config_path(args.osv_dir or get_required_value(config, "clone", "osv_dir"))
+        target_dir = resolve_config_path(args.target_dir or get_required_value(config, "clone", "target_dir"))
     except ValueError as error:
         parser.error(f"{error} (config: {config_source})")
+    max_repos = args.max_repos
+    if max_repos is None:
+        max_repos_str = config.get("max_repos", fallback="").strip()
+        if max_repos_str:
+            try:
+                max_repos = int(max_repos_str)
+            except ValueError as error:
+                parser.error(f"Invalid clone.max_repos value '{max_repos_str}' in {config_source}: {error}")
     target_dir.mkdir(parents=True, exist_ok=True)
 
     repo_urls = set()
@@ -76,8 +89,8 @@ def main() -> None:
         repo_urls.update(extract_repo_urls(vulnerability))
 
     ordered_repos = sorted(repo_urls)
-    if args.max_repos is not None:
-        ordered_repos = ordered_repos[: args.max_repos]
+    if max_repos is not None:
+        ordered_repos = ordered_repos[:max_repos]
 
     for repo_url in ordered_repos:
         clone_or_update(repo_url, target_dir, args.update_existing)

scripts/enrich_osv_recidivism.py

@@ -8,7 +8,7 @@
 from urllib.request import urlretrieve
 
 from osv_common import collect_history, iter_vulnerability_files, load_vulnerability, recidivism_for_vulnerability
-from recidivism_config import load_config_with_source, required_value, resolve_config_path
+from recidivism_config import get_required_value, load_config_with_source, resolve_config_path
 
 
 def download_dump(url: str, destination: Path, force: bool) -> None:
@@ -37,10 +37,10 @@ def main() -> None:
     config, config_source = load_config_with_source("enrich")
 
     parser = argparse.ArgumentParser(description="Download OSV dump and enrich with recidivism metrics.")
-    parser.add_argument("--dump-url")
-    parser.add_argument("--archive-path")
-    parser.add_argument("--extract-dir")
-    parser.add_argument("--output")
+    parser.add_argument("--dump-url", help="Override enrich.dump_url from recidivism.ini")
+    parser.add_argument("--archive-path", help="Override enrich.archive_path from recidivism.ini")
+    parser.add_argument("--extract-dir", help="Override enrich.extract_dir from recidivism.ini")
+    parser.add_argument("--output", help="Override enrich.output from recidivism.ini")
     parser.add_argument(
         "--force-download",
         action=argparse.BooleanOptionalAction,
@@ -54,10 +54,10 @@ def main() -> None:
     args = parser.parse_args()
 
     try:
-        dump_url = args.dump_url or required_value(config, "dump_url")
-        archive_path = resolve_config_path(args.archive_path or required_value(config, "archive_path"))
-        extract_dir = resolve_config_path(args.extract_dir or required_value(config, "extract_dir"))
-        output_path = resolve_config_path(args.output or required_value(config, "output"))
+        dump_url = args.dump_url or get_required_value(config, "enrich", "dump_url")
+        archive_path = resolve_config_path(args.archive_path or get_required_value(config, "enrich", "archive_path"))
+        extract_dir = resolve_config_path(args.extract_dir or get_required_value(config, "enrich", "extract_dir"))
+        output_path = resolve_config_path(args.output or get_required_value(config, "enrich", "output"))
     except ValueError as error:
         parser.error(f"{error} (config: {config_source})")
 

scripts/recidivism_config.py

@@ -65,11 +65,12 @@ def resolve_config_path(path_value: str) -> Path:
     return (REPO_ROOT / path).resolve()
 
 
-def required_value(config: configparser.SectionProxy, key: str) -> str:
+def get_required_value(config: configparser.SectionProxy, section: str, key: str) -> str:
     """Return a required non-empty configuration value.
 
     Args:
         config: Configuration section containing script settings.
+        section: Section name for diagnostics.
         key: Config key to read.
 
     Returns:
@@ -78,7 +79,7 @@ def required_value(config: configparser.SectionProxy, key: str) -> str:
     Raises:
         ValueError: If the key is missing or empty.
     """
-    value = config.get(key)
+    value = config.get(key, fallback=None)
     if value is None or not value.strip():
-        raise ValueError(f"Missing required config key '{key}' in section [{config.name}].")
+        raise ValueError(f"Missing required config key '{key}' in section [{section}].")
     return value

tests/test_recidivism_config.py

@@ -8,11 +8,11 @@
 
 sys.path.insert(0, str((Path(__file__).resolve().parents[1] / "scripts")))
 
-from recidivism_config import required_value, load_config, resolve_config_path  # noqa: E402
+from recidivism_config import get_required_value, load_config, resolve_config_path  # noqa: E402
 
 
 class RecidivismConfigTests(unittest.TestCase):
-    def test_loads_default_and_prints_message_when_local_missing(self) -> None:
+    def test_fallback_to_default_when_local_missing(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             tmp_path = Path(tmp)
             default_config = tmp_path / "recidivism.default.ini"
@@ -43,7 +43,7 @@ def test_required_value_raises_for_empty(self) -> None:
             with patch("recidivism_config.LOCAL_CONFIG_FILE", config_path):
                 section = load_config("clone")
             with self.assertRaises(ValueError):
-                required_value(section, "osv_dir")
+                get_required_value(section, "clone", "osv_dir")
 
 
 if __name__ == "__main__":