feat: make PoP (IAL-1) the default badge mode with CA fallback (#69)

* feat: make PoP (IAL-1) the default badge mode with CA fallback

CapiscIO.connect() now tries Proof-of-Possession badge issuance first:
1. Reads the agent's private key from keys_dir
2. Requests a PoP challenge from the registry
3. Signs the challenge to prove key ownership
4. Receives an IAL-1 badge (higher assurance than CA-issued)

Falls back to CA-issued badge (IAL-0) via the gRPC keeper if PoP fails
(e.g. server doesn't support PoP yet, network issues).

Also passes private_key_path through to BadgeKeeper so the Go core can
find the key in custom keys_dir locations (not just ~/.capiscio/keys/).

* fix: prevent keeper from overwriting PoP badge + handle nil jti

- Only start BadgeKeeper when PoP fails (CA fallback path)
- Prevent IAL-1 PoP badge from being overwritten by IAL-0 CA badge
- Only pass private_key_path when file exists (ephemeral env safety)
- Use null-coalescing for jti to prevent TypeError on None values

Addresses review feedback from PR #69.

* fix: rename CAPISCIO_KEYS_DIR → CAPISCIO_KEY_DIR per documented convention

Author: beonde

capiscio_sdk/badge_keeper.py

@@ -56,6 +56,7 @@ class BadgeKeeperConfig:
         check_interval: Check interval in seconds (default: 5)
         trust_level: Trust level for CA mode (1-4, default: 1)
         rpc_address: Optional custom RPC address for capiscio-core
+        private_key_path: Optional path to private key JWK file
         on_renew: Optional callback(token: str) called when badge renews
         max_retries: Max retry attempts on renewal failure (default: 3)
         retry_backoff: Base backoff seconds for exponential retry (default: 2)
@@ -70,6 +71,7 @@ class BadgeKeeperConfig:
     check_interval: int = 5
     trust_level: int = 1
     rpc_address: Optional[str] = None
+    private_key_path: Optional[str] = None
     on_renew: Optional[Callable[[str], None]] = None
     max_retries: int = 3
     retry_backoff: int = 2
@@ -98,6 +100,7 @@ def __init__(
         check_interval: int = 5,
         trust_level: int = 1,
         rpc_address: Optional[str] = None,
+        private_key_path: Optional[str] = None,
         on_renew: Optional[Callable[[str], None]] = None,
         max_retries: int = 3,
         retry_backoff: int = 2,
@@ -115,6 +118,7 @@ def __init__(
             check_interval: Check interval in seconds (default: 5)
             trust_level: Trust level for CA mode (1-4, default: 1)
             rpc_address: Optional custom RPC address for capiscio-core
+            private_key_path: Optional path to private key JWK file
             on_renew: Optional callback(token: str) called when badge renews
             max_retries: Max retry attempts on renewal failure (default: 3)
             retry_backoff: Base backoff seconds for exponential retry (default: 2)
@@ -130,6 +134,7 @@ def __init__(
             check_interval=check_interval,
             trust_level=trust_level,
             rpc_address=rpc_address,
+            private_key_path=private_key_path,
             on_renew=on_renew,
             max_retries=max_retries,
             retry_backoff=retry_backoff,
@@ -222,6 +227,7 @@ def _run_keeper(self) -> None:
                 renew_before_seconds=self.config.renewal_threshold,
                 check_interval_seconds=self.config.check_interval,
                 trust_level=self.config.trust_level,
+                private_key_path=self.config.private_key_path or "",
             ):
                 # Check stop signal
                 if self._stop_event.is_set():

capiscio_sdk/connect.py

@@ -42,6 +42,9 @@
 # Env var for injecting the private key in ephemeral environments
 ENV_AGENT_PRIVATE_KEY = "CAPISCIO_AGENT_PRIVATE_KEY_JWK"
 
+# Env var for overriding the keys directory
+ENV_KEYS_DIR = "CAPISCIO_KEY_DIR"
+
 
 # =============================================================================
 # Key injection helpers
@@ -223,7 +226,9 @@ def emit(self, event_type: str, data: Dict[str, Any]) -> bool:
     def get_badge(self) -> Optional[str]:
         """Get current badge (auto-renewed if needed)."""
         if self._keeper:
-            return self._keeper.get_current_badge()
+            badge = self._keeper.get_current_badge()
+            if badge:
+                return badge
         return self.badge
 
     def status(self) -> Dict[str, Any]:
@@ -335,6 +340,10 @@ def connect(
             agent_id = os.environ.get("CAPISCIO_AGENT_ID")
         if server_url is None:
             server_url = os.environ.get("CAPISCIO_SERVER_URL") or PROD_REGISTRY
+        if keys_dir is None:
+            env_keys = os.environ.get(ENV_KEYS_DIR)
+            if env_keys:
+                keys_dir = Path(env_keys)
 
         connector = _Connector(
             api_key=api_key,
@@ -870,7 +879,15 @@ def _activate_agent(self):
             logger.debug(f"Agent activation failed: {e} - non-critical")
 
     def _setup_badge(self):
-        """Set up BadgeKeeper for automatic badge management."""
+        """Set up badge management with PoP (IAL-1) preferred, CA (IAL-0) fallback.
+        
+        Per RFC-003, agents SHOULD prove key possession to obtain IAL-1
+        badges. This method tries PoP first, falling back to CA if the
+        server or agent keys are not PoP-ready.
+        
+        The keeper runs in CA mode for continuous renewal because the
+        keeper streaming RPC does not yet support PoP mode.
+        """
         try:
             from .badge_keeper import BadgeKeeper
             from .simple_guard import SimpleGuard
@@ -884,34 +901,97 @@ def _setup_badge(self):
                 keys_preloaded=True,
             )
 
-            # Set up BadgeKeeper with correct parameters
-            # Wire on_renew so the guard's badge token stays in sync
-            # when the keeper renews it in the background.
-            keeper = BadgeKeeper(
-                api_url=self.server_url,
-                api_key=self.api_key,
-                agent_id=self.agent_id,
-                mode="dev" if self.dev_mode else "ca",
-                output_file=str(self.keys_dir / "badge.jwt"),
-                on_renew=lambda token: guard.set_badge_token(token),
-            )
-            
-            # Start the keeper and get initial badge
-            keeper.start()
-            badge = keeper.get_current_badge()
-            # Get expiration from keeper if available, otherwise None
+            badge = None
             expires_at = None
-            if hasattr(keeper, 'badge_expires_at'):
-                expires_at = keeper.badge_expires_at
-            elif hasattr(keeper, 'get_badge_expiration'):
-                expires_at = keeper.get_badge_expiration()
+            
+            # Try PoP (IAL-1) for the initial badge — secure by default (RFC-003)
+            if not self.dev_mode:
+                badge, expires_at = self._request_pop_badge(guard)
+            
+            # Start keeper for continuous renewal (CA mode).
+            # Only start if PoP didn't succeed — otherwise the keeper would
+            # immediately overwrite the IAL-1 PoP badge with an IAL-0 CA badge.
+            # When PoP-based renewal is supported, keeper can be started always.
+            keeper = None
+            if badge is None:
+                private_key_file = self.keys_dir / "private.jwk"
+                keeper = BadgeKeeper(
+                    api_url=self.server_url,
+                    api_key=self.api_key,
+                    agent_id=self.agent_id,
+                    mode="dev" if self.dev_mode else "ca",
+                    output_file=str(self.keys_dir / "badge.jwt"),
+                    private_key_path=str(private_key_file) if private_key_file.exists() else None,
+                    on_renew=lambda token: guard.set_badge_token(token),
+                )
+                keeper.start()
+                badge = keeper.get_current_badge()
+                if hasattr(keeper, 'badge_expires_at'):
+                    expires_at = keeper.badge_expires_at
+                elif hasattr(keeper, 'get_badge_expiration'):
+                    expires_at = keeper.get_badge_expiration()
 
             return badge, expires_at, keeper, guard
 
         except Exception as e:
             logger.warning(f"Badge setup failed (continuing without badge): {e}")
             return None, None, None, None
 
+    def _request_pop_badge(self, guard):
+        """Request initial badge via PoP challenge-response (RFC-003).
+        
+        Returns an IAL-1 badge with cryptographic key binding (cnf claim).
+        Falls back gracefully if PoP is unavailable.
+        """
+        try:
+            private_key_path = self.keys_dir / "private.jwk"
+            if not private_key_path.exists():
+                logger.warning(
+                    "No private key at %s — skipping PoP, will use CA (IAL-0)",
+                    private_key_path,
+                )
+                return None, None
+            
+            private_key_jwk = private_key_path.read_text(encoding="utf-8").strip()
+            
+            # Ensure RPC client is available (identity recovery path
+            # may not have created one)
+            if self._rpc_client is None:
+                self._rpc_client = CapiscioRPCClient()
+                self._rpc_client.connect()
+            
+            success, result, error = self._rpc_client.badge.request_pop_badge(
+                agent_did=self.did,
+                private_key_jwk=private_key_jwk,
+                api_key=self.api_key,
+                ca_url=self.server_url,
+            )
+            
+            if success and result:
+                token = result["token"]
+                guard.set_badge_token(token)
+                
+                # Persist badge to disk
+                badge_path = self.keys_dir / "badge.jwt"
+                badge_path.write_text(token, encoding="utf-8")
+                
+                logger.info(
+                    "PoP badge acquired (IAL-1, jti=%s...)",
+                    (result.get("jti") or "unknown")[:8],
+                )
+                return token, result.get("expires_at")
+            else:
+                logger.warning(
+                    "PoP badge request failed: %s — falling back to CA (IAL-0)",
+                    error or "unknown error",
+                )
+                return None, None
+        except Exception as e:
+            logger.warning(
+                "PoP badge request error: %s — falling back to CA (IAL-0)", e
+            )
+            return None, None
+
 
 # Convenience alias
 connect = CapiscIO.connect