[#2273 Items A,B,C] Ed25519 key rotation, registry expiry, and non-root fallback

node/p2p_identity.py

@@ -49,16 +49,44 @@
 SIGNING_MODE = _MODE_RAW
 
 # Paths
-DEFAULT_PRIVKEY_PATH = os.environ.get(
-    "RC_P2P_PRIVKEY_PATH",
-    "/etc/rustchain/p2p_identity.pem",
-)
+DEFAULT_PRIVKEY_PATH = "/etc/rustchain/p2p_identity.pem"
 DEFAULT_REGISTRY_PATH = os.environ.get(
     "RC_P2P_PEER_REGISTRY",
     "/etc/rustchain/peer_registry.json",
 )
 
 
+def get_default_privkey_path() -> Path:
+    """Return the first writable private key path in priority order."""
+    env_path = os.environ.get("RC_P2P_PRIVKEY_PATH")
+    if env_path:
+        return Path(env_path)
+
+    paths = [
+        Path("/etc/rustchain/p2p_identity.pem"),
+        Path.home() / ".rustchain" / "p2p_identity.pem",
+    ]
+
+    # Use the first one that exists
+    for p in paths:
+        if p.exists():
+            return p
+
+    # Otherwise, return the first one we can write to (or the last fallback)
+    for p in paths:
+        try:
+            p.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+            # Try to create/append to a dummy file to check writability
+            test_file = p.parent / ".write_test"
+            test_file.touch()
+            test_file.unlink()
+            return p
+        except (PermissionError, OSError):
+            continue
+
+    return paths[-1]
+
+
 # ---------------------------------------------------------------------------
 # Optional dependency: cryptography.
 #
@@ -105,8 +133,12 @@ class LocalKeypair:
     file. Public key is exposed as hex.
     """
 
-    def __init__(self, path: str = DEFAULT_PRIVKEY_PATH):
-        self.path = Path(path)
+    def __init__(self, path: Optional[str | Path] = None):
+        if path is None:
+            self.path = get_default_privkey_path()
+        else:
+            self.path = Path(path)
+        self.key_version = 1  # Item A: key rotation
         self._privkey = None  # lazy
         self._pubkey_hex: Optional[str] = None
 
@@ -121,23 +153,53 @@ def _load_or_generate(self):
             _InvalidSignature,
         ) = _require_crypto()
 
-        if self.path.exists():
+        # Item A: Look for versioned key file if forced or if current exists
+        force_keygen = os.environ.get("RC_P2P_KEYGEN", "0") == "1"
+
+        if self.path.exists() and not force_keygen:
             with open(self.path, "rb") as f:
-                self._privkey = load_pem_private_key(f.read(), password=None)
-            logger.info(f"[P2P] Loaded Ed25519 identity from {self.path}")
+                content = f.read()
+                self._privkey = load_pem_private_key(content, password=None)
+                version_path = self.path.with_suffix(".version")
+                if version_path.exists():
+                    try:
+                        self.key_version = int(version_path.read_text().strip())
+                    except ValueError:
+                        self.key_version = 1
+            logger.info(f"[P2P] Loaded Ed25519 identity (v{self.key_version}) from {self.path}")
         else:
+            if force_keygen and self.path.exists():
+                # Item A: keep old keypair for rollback grace
+                version_path = self.path.with_suffix(".version")
+                current_v = 1
+                if version_path.exists():
+                    try:
+                        current_v = int(version_path.read_text().strip())
+                    except ValueError:
+                        pass
+
+                old_path = self.path.parent / f"{self.path.stem}.v{current_v}.pem"
+                self.path.replace(old_path)
+                logger.info(f"[P2P] Archived old identity to {old_path}")
+                self.key_version = current_v + 1
+
             self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
             self._privkey = Ed25519PrivateKey.generate()
             pem = self._privkey.private_bytes(
                 Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()
             )
             # Write with 0600 perms
-            fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+            fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
             try:
                 os.write(fd, pem)
             finally:
                 os.close(fd)
-            logger.info(f"[P2P] Generated new Ed25519 identity at {self.path}")
+
+            # Persist version
+            version_path = self.path.with_suffix(".version")
+            version_path.write_text(str(self.key_version))
+
+            logger.info(f"[P2P] Generated new Ed25519 identity (v{self.key_version}) at {self.path}")
 
         from cryptography.hazmat.primitives.serialization import (
             Encoding as _Enc,
@@ -166,6 +228,9 @@ def pubkey_hex(self) -> str:
 class PeerEntry:
     node_id: str
     pubkey_hex: str
+    key_version: int = 1
+    not_before: Optional[str] = None  # ISO-8601
+    not_after: Optional[str] = None   # ISO-8601
 
 
 class PeerRegistry:
@@ -175,7 +240,13 @@ class PeerRegistry:
         {
           "version": 1,
           "peers": [
-            {"node_id": "...", "pubkey_hex": "..."},
+            {
+              "node_id": "...",
+              "pubkey_hex": "...",
+              "key_version": 1,
+              "not_before": "2026-04-01T00:00:00Z",
+              "not_after": "2027-04-01T00:00:00Z"
+            },
             ...
           ],
           "cluster_root_sig": "..."    # optional root-signed attestation
@@ -203,10 +274,19 @@ def load(self) -> None:
         for p in peers:
             nid = p.get("node_id")
             pk = p.get("pubkey_hex")
+            kv = p.get("key_version", 1)
+            nb = p.get("not_before")
+            na = p.get("not_after")
             if not nid or not pk:
                 logger.warning(f"[P2P] Skipping malformed peer entry: {p}")
                 continue
-            entries[nid] = PeerEntry(node_id=nid, pubkey_hex=pk)
+            entries[nid] = PeerEntry(
+                node_id=nid,
+                pubkey_hex=pk,
+                key_version=kv,
+                not_before=nb,
+                not_after=na
+            )
         self._by_node_id = entries
         self._loaded = True
         logger.info(f"[P2P] Loaded {len(entries)} peers from registry {self.path}")
@@ -215,7 +295,43 @@ def get_pubkey(self, node_id: str) -> Optional[str]:
         if not self._loaded:
             self.load()
         entry = self._by_node_id.get(node_id)
-        return entry.pubkey_hex if entry else None
+        if not entry:
+            return None
+
+        # Item B: Registry expiry / not_before / not_after
+        from datetime import datetime, timezone
+        now = datetime.now(timezone.utc)
+
+        # Clock skew tolerance: ±5 min (300s)
+        SKEW = 300
+
+        if entry.not_before:
+            try:
+                nb = datetime.fromisoformat(entry.not_before.replace("Z", "+00:00"))
+                if (nb.timestamp() - now.timestamp()) > SKEW:
+                    logger.warning(f"[P2P] Peer {node_id} registry entry not yet valid (not_before={entry.not_before})")
+                    return None
+            except ValueError:
+                logger.warning(f"[P2P] Peer {node_id} has invalid not_before: {entry.not_before}")
+
+        if entry.not_after:
+            try:
+                na = datetime.fromisoformat(entry.not_after.replace("Z", "+00:00"))
+                if (now.timestamp() - na.timestamp()) > SKEW:
+                    logger.warning(f"[P2P] Peer {node_id} registry entry expired (not_after={entry.not_after})")
+                    return None
+            except ValueError:
+                logger.warning(f"[P2P] Peer {node_id} has invalid not_after: {entry.not_after}")
+
+        return entry.pubkey_hex
+
+    def get_entry(self, node_id: str) -> Optional[PeerEntry]:
+        if not self._loaded:
+            self.load()
+        # Returns pubkey if valid per get_pubkey, then the entry object
+        if self.get_pubkey(node_id) is None:
+            return None
+        return self._by_node_id.get(node_id)
 
     def __len__(self) -> int:
         if not self._loaded:
@@ -226,7 +342,7 @@ def __len__(self) -> int:
 # ---------------------------------------------------------------------------
 # Signature bundle: JSON-encoded dual signature (or legacy hex)
 # ---------------------------------------------------------------------------
-def pack_signature(hmac_sig: Optional[str], ed25519_sig: Optional[str]) -> str:
+def pack_signature(hmac_sig: Optional[str], ed25519_sig: Optional[str], key_version: int = 1) -> str:
     """Pack one or two signatures into the wire-format signature field.
 
     - HMAC only (legacy): return hex string as-is.
@@ -238,26 +354,27 @@ def pack_signature(hmac_sig: Optional[str], ed25519_sig: Optional[str]) -> str:
     if hmac_sig:
         bundle["h"] = hmac_sig
     bundle["e"] = ed25519_sig
+    bundle["v"] = key_version
     return json.dumps(bundle, separators=(",", ":"))
 
 
-def unpack_signature(sig_field: str) -> Tuple[Optional[str], Optional[str]]:
+def unpack_signature(sig_field: str) -> Tuple[Optional[str], Optional[str], int]:
     """Inverse of pack_signature.
 
-    Returns (hmac_sig, ed25519_sig). Either may be None if not present.
-    Treats raw-hex strings as legacy HMAC-only.
+    Returns (hmac_sig, ed25519_sig, key_version). Either sig may be None if not present.
+    Treats raw-hex strings as legacy HMAC-only with version 1.
     """
     if not sig_field:
-        return None, None
+        return None, None, 1
     stripped = sig_field.strip()
     if stripped.startswith("{"):
         try:
             bundle = json.loads(stripped)
-            return bundle.get("h"), bundle.get("e")
+            return bundle.get("h"), bundle.get("e"), bundle.get("v", 1)
         except json.JSONDecodeError:
-            return None, None
-    # Legacy hex — assume HMAC
-    return stripped, None
+            return None, None, 1
+    # Legacy hex — assume HMAC, version 1
+    return stripped, None, 1
 
 
 # ---------------------------------------------------------------------------

node/tests/test_non_root_path.py

@@ -0,0 +1,49 @@
+import unittest
+import os
+import shutil
+import tempfile
+from pathlib import Path
+from node.p2p_identity import get_default_privkey_path, LocalKeypair
+
+class TestNonRootKeyPath(unittest.TestCase):
+    def setUp(self):
+        self.tmp_home = tempfile.mkdtemp()
+        self.old_home = os.environ.get("HOME")
+        os.environ["HOME"] = self.tmp_home
+
+        # Ensure /etc/rustchain is "unwritable" (doesn't exist in our environment usually, 
+        # or we can mock it if needed. For now let's assume it fails and we hit HOME.)
+
+    def tearDown(self):
+        if self.old_home:
+            os.environ["HOME"] = self.old_home
+        shutil.rmtree(self.tmp_home)
+
+    def test_fallback_to_user_home(self):
+        """Assert that if /etc/rustchain is unwritable, we fall back to $HOME/.rustchain"""
+        # We assume /etc/rustchain/p2p_identity.pem is unwritable for the current user (albega)
+        # unless albega is root.
+
+        path = get_default_privkey_path()
+
+        # In a normal non-root environment, it should be the HOME one
+        expected_home_path = Path(self.tmp_home) / ".rustchain" / "p2p_identity.pem"
+
+        # Note: on some CI it might actually be writable. Let's check.
+        if os.access("/etc", os.W_OK):
+             # Skip or adjust if we are actually root
+             self.skipTest("Running as root/sudo, /etc is writable.")
+
+        self.assertEqual(path, expected_home_path)
+
+    def test_local_keypair_uses_fallback(self):
+        """Assert LocalKeypair uses the fallback path automatically."""
+        if os.access("/etc", os.W_OK):
+             self.skipTest("Running as root/sudo, /etc is writable.")
+
+        kp = LocalKeypair()
+        expected_home_path = Path(self.tmp_home) / ".rustchain" / "p2p_identity.pem"
+        self.assertEqual(kp.path, expected_home_path)
+
+if __name__ == "__main__":
+    unittest.main()