feat: add engine identity module (FQDN resolution via node.env, hostname fallback)

engine/identity.py

@@ -0,0 +1,76 @@
+# engine/identity.py
+"""Node identity from host FQDN + node.env (replaces bash root_require_node).
+
+Decision B: the FQDN <node>.a<x>.apex.ermnvldmr.com keeps the cluster; we parse
+node+cluster from it. node.env owns APEX_SUBNET and mirrors APEX_CLUSTER; hostname
+wins on drift. Falls back to the repo directory name when the FQDN is not yet
+apex-shaped (mid-cutover).
+"""
+from __future__ import annotations
+
+import os
+import re
+import sys
+from dataclasses import dataclass
+from typing import Dict, List, Tuple
+
+_FQDN = re.compile(r"^(?P<node>[^.]+)\.(?P<cluster>a\d+)\.apex\.ermnvldmr\.com$")
+_DIRNAME = re.compile(r"^(?P<node>[^.]+)\.apex\.ermnvldmr\.com")
+
+
+@dataclass
+class Node:
+    name: str
+    cluster: str
+    subnet: str
+    fqdn: str
+
+
+def read_env(path: str) -> Dict[str, str]:
+    """Parse a KEY=value file (shell-sourceable). Ignores comments/blank lines."""
+    out: Dict[str, str] = {}
+    if not os.path.isfile(path):
+        return out
+    with open(path) as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, _, val = line.partition("=")
+            out[key.strip()] = val.strip().strip('"').strip("'")
+    return out
+
+
+def resolve(fqdn: str, repo_root: str) -> Tuple[Node, List[str]]:
+    """Return (Node, warnings). Exits 66 (EX_NOINPUT) if APEX_SUBNET is missing."""
+    warns: List[str] = []
+    env = read_env(os.path.join(repo_root, "node.env"))
+
+    m = _FQDN.match(fqdn or "")
+    if m:
+        name, cluster = m.group("node"), m.group("cluster")
+        env_cluster = env.get("APEX_CLUSTER")
+        if env_cluster and env_cluster != cluster:
+            warns.append(
+                f"cluster drift: hostname says {cluster!r}, node.env says "
+                f"{env_cluster!r}; hostname wins."
+            )
+    else:
+        # Fallback: derive node from repo dir name; cluster from node.env.
+        warns.append(
+            f"FQDN {fqdn!r} is not apex-shaped; falling back to repo dir + node.env."
+        )
+        base = os.path.basename(os.path.realpath(repo_root))
+        dm = _DIRNAME.match(base)
+        name = dm.group("node") if dm else base
+        cluster = env.get("APEX_CLUSTER", "")
+        if not cluster:
+            sys.stderr.write("identity: no cluster in FQDN or node.env\n")
+            raise SystemExit(66)
+
+    subnet = env.get("APEX_SUBNET", "")
+    if not subnet:
+        sys.stderr.write("identity: APEX_SUBNET missing from node.env\n")
+        raise SystemExit(66)
+
+    return Node(name=name, cluster=cluster, subnet=subnet, fqdn=fqdn), warns

tests/test_identity.py

@@ -0,0 +1,42 @@
+# tests/test_identity.py
+import os, tempfile, unittest
+from engine.identity import resolve, Node
+
+class TestIdentity(unittest.TestCase):
+    def _envfile(self, text, dirname=None):
+        d = tempfile.mkdtemp()
+        if dirname:
+            d = os.path.join(d, dirname)
+            os.makedirs(d)
+        with open(os.path.join(d, "node.env"), "w") as f:
+            f.write(text)
+        return d
+
+    def test_fqdn_is_source_of_truth(self):
+        repo = self._envfile("APEX_CLUSTER=a1\nAPEX_SUBNET=198.18.11.0/24\n")
+        n, warns = resolve("daedalus.a1.apex.ermnvldmr.com", repo)
+        self.assertEqual((n.name, n.cluster, n.subnet), ("daedalus", "a1", "198.18.11.0/24"))
+        self.assertEqual(warns, [])
+
+    def test_cluster_drift_warns_hostname_wins(self):
+        repo = self._envfile("APEX_CLUSTER=a9\nAPEX_SUBNET=198.18.11.0/24\n")
+        n, warns = resolve("daedalus.a1.apex.ermnvldmr.com", repo)
+        self.assertEqual(n.cluster, "a1")           # hostname wins
+        self.assertTrue(any("drift" in w.lower() for w in warns))
+
+    def test_fallback_when_fqdn_not_apex_shaped(self):
+        # repo dir basename carries the node name in fallback (real host layout:
+        # $HOME/morpheus.apex.ermnvldmr.com)
+        repo = self._envfile("APEX_CLUSTER=a2\nAPEX_SUBNET=198.18.13.0/24\n",
+                             dirname="morpheus.apex.ermnvldmr.com")
+        n, warns = resolve("build-laptop", repo)
+        self.assertEqual((n.name, n.cluster, n.subnet), ("morpheus", "a2", "198.18.13.0/24"))
+        self.assertTrue(any("fqdn" in w.lower() for w in warns))
+
+    def test_missing_subnet_is_error(self):
+        repo = self._envfile("APEX_CLUSTER=a1\n")
+        with self.assertRaises(SystemExit):
+            resolve("daedalus.a1.apex.ermnvldmr.com", repo)
+
+if __name__ == "__main__":
+    unittest.main()