fix(otdf-local): address coderabbit review feedback

- cli_scenario: set OTDF_LOCAL_INSTANCE_NAME + clear settings cache before
  get_settings() so scenario-driven instance name is picked up
- cli_instance: add _validate_instance_name() to guard against path traversal
  in init/rm; add --force flag to init to prevent silent overwrite
- kas: add get_instance_names() public method; replace _instances access in cli
- keys: generate_ca_jks() now imports cert only (keytool -importcert) so ca.jks
  is a proper truststore; ensure_keys_exist() guards include cert files alongside
  private keys to catch partial-init broken state

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dmihalcik-virtru

otdf-local/src/otdf_local/cli.py

@@ -220,7 +220,7 @@ def up(
             raise typer.Exit(1)
 
         with status_spinner("Waiting for KAS instances..."):
-            for kas_name in kas_manager._instances:
+            for kas_name in kas_manager.get_instance_names():
                 port = settings.get_kas_port(kas_name)
                 try:
                     wait_for_health(

otdf-local/src/otdf_local/cli_instance.py

@@ -22,6 +22,17 @@
 instance_app = typer.Typer(help="Manage named test environment instances.")
 
 
+def _validate_instance_name(name: str) -> None:
+    """Reject names that could escape the instances root via path traversal."""
+    from pathlib import PurePosixPath
+
+    p = PurePosixPath(name)
+    if not name or p.is_absolute() or len(p.parts) != 1 or name in {".", ".."}:
+        raise typer.BadParameter(
+            f"instance name must be a single directory name, got {name!r}"
+        )
+
+
 @instance_app.command("init")
 def init(
     name: Annotated[str, typer.Argument(help="Instance name (used as directory name)")],
@@ -41,11 +52,24 @@ def init(
         Optional[str],
         typer.Option("--platform", help="Platform dist version (e.g., v0.9.0)"),
     ] = None,
+    force: Annotated[
+        bool,
+        typer.Option("--force", help="Overwrite existing instance directory"),
+    ] = False,
 ) -> None:
     """Scaffold a new instance directory at tests/instances/<name>/."""
+    _validate_instance_name(name)
     settings = get_settings()
     instance_dir = settings.instances_root / name
 
+    if instance_dir.exists() and not force:
+        typer.echo(
+            f"Error: instance '{name}' already exists at {instance_dir}. "
+            "Pass --force to overwrite.",
+            err=True,
+        )
+        raise typer.Exit(2)
+
     if from_scenario is not None:
         _init_from_scenario(name, from_scenario, instance_dir)
     else:
@@ -263,6 +287,7 @@ def rm(
     yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
 ) -> None:
     """Remove an instance directory."""
+    _validate_instance_name(name)
     settings = get_settings()
     instance_dir = settings.instances_root / name
     if not instance_dir.exists():

otdf-local/src/otdf_local/cli_scenario.py

@@ -80,9 +80,10 @@ def run(
         )
         raise typer.Exit(2)
 
-    settings = get_settings()
     # Force the chosen instance via env so child pytest invocations agree.
     os.environ["OTDF_LOCAL_INSTANCE_NAME"] = instance_name
+    get_settings.cache_clear()
+    settings = get_settings()
 
     xtest_root = settings.xtest_root
     if not xtest_root.exists():

otdf-local/src/otdf_local/services/kas.py

@@ -249,6 +249,10 @@ def get_all_info(self) -> list[ServiceInfo]:
         """Get info for all KAS instances."""
         return [instance.get_info() for instance in self._instances.values()]
 
+    def get_instance_names(self) -> list[str]:
+        """Return names of all managed KAS instances."""
+        return list(self._instances.keys())
+
     def get_running(self) -> list[str]:
         """Get names of running KAS instances."""
         return [name for name, inst in self._instances.items() if inst.is_running()]

otdf-local/src/otdf_local/utils/keys.py

@@ -234,41 +234,23 @@ def generate_localhost_cert(key_dir: Path) -> tuple[Path, Path]:
 
 
 def generate_ca_jks(key_dir: Path, password: str = "password") -> Path:
-    """Convert the keycloak CA into the JKS truststore Keycloak mounts.
+    """Convert the keycloak CA certificate into a JKS truststore Keycloak mounts.
 
     Uses keytool inside the keycloak/keycloak:25.0 image so we don't need a
     local JDK — docker is already a hard dependency for the test env.
     Requires generate_localhost_cert() to have run first.
+
+    Only the CA certificate (public) is imported — not the private key — so the
+    JKS is a proper truststore, not a keystore.
     """
-    ca_key = key_dir / "keycloak-ca-private.pem"
     ca_cert = key_dir / "keycloak-ca.pem"
-    if not ca_key.exists() or not ca_cert.exists():
+    if not ca_cert.exists():
         raise FileNotFoundError(
-            f"CA files missing in {key_dir}; call generate_localhost_cert() first"
+            f"CA certificate missing in {key_dir}; call generate_localhost_cert() first"
         )
-    p12 = key_dir / "ca.p12"
     jks = key_dir / "ca.jks"
 
-    subprocess.run(
-        [
-            "openssl",
-            "pkcs12",
-            "-export",
-            "-in",
-            str(ca_cert),
-            "-inkey",
-            str(ca_key),
-            "-out",
-            str(p12),
-            "-nodes",
-            "-passout",
-            f"pass:{password}",
-        ],
-        check=True,
-        capture_output=True,
-    )
-
-    # keytool -importkeystore via the keycloak image (matches init-temp-keys.sh)
+    # keytool -importcert via the keycloak image: cert-only truststore entry
     result = subprocess.run(
         [
             "docker",
@@ -281,18 +263,16 @@ def generate_ca_jks(key_dir: Path, password: str = "password") -> Path:
             "--user",
             f"{os.getuid()}:{os.getgid()}",
             "keycloak/keycloak:25.0",
-            "-importkeystore",
-            "-srckeystore",
-            "/keys/ca.p12",
-            "-srcstoretype",
-            "PKCS12",
-            "-destkeystore",
+            "-importcert",
+            "-file",
+            "/keys/keycloak-ca.pem",
+            "-alias",
+            "ca",
+            "-keystore",
             "/keys/ca.jks",
-            "-deststoretype",
+            "-storetype",
             "JKS",
-            "-srcstorepass",
-            password,
-            "-deststorepass",
+            "-storepass",
             password,
             "-noprompt",
         ],
@@ -301,7 +281,7 @@ def generate_ca_jks(key_dir: Path, password: str = "password") -> Path:
     )
     if result.returncode != 0:
         raise RuntimeError(
-            f"keytool failed converting PKCS12 → JKS:\n{result.stderr}\n"
+            f"keytool failed importing CA cert into JKS truststore:\n{result.stderr}\n"
             "Ensure Docker is running and `keycloak/keycloak:25.0` is pullable."
         )
     return jks
@@ -323,24 +303,30 @@ def ensure_keys_exist(key_dir: Path, force: bool = False) -> bool:
         True if any keys were generated, False if everything already existed
     """
     rsa_private = key_dir / "kas-private.pem"
+    rsa_cert = key_dir / "kas-cert.pem"
     ec_private = key_dir / "kas-ec-private.pem"
+    ec_cert = key_dir / "kas-ec-cert.pem"
     localhost_key = key_dir / "localhost.key"
+    localhost_cert = key_dir / "localhost.crt"
     ca_jks = key_dir / "ca.jks"
 
     if (
         not force
         and rsa_private.exists()
+        and rsa_cert.exists()
         and ec_private.exists()
+        and ec_cert.exists()
         and localhost_key.exists()
+        and localhost_cert.exists()
         and ca_jks.exists()
     ):
         return False
 
-    if force or not rsa_private.exists():
+    if force or not rsa_private.exists() or not rsa_cert.exists():
         generate_rsa_keypair(key_dir, "kas")
-    if force or not ec_private.exists():
+    if force or not ec_private.exists() or not ec_cert.exists():
         generate_ec_keypair(key_dir, "kas-ec")
-    if force or not localhost_key.exists():
+    if force or not localhost_key.exists() or not localhost_cert.exists():
         generate_localhost_cert(key_dir)
     if force or not ca_jks.exists():
         generate_ca_jks(key_dir)