1  #!/usr/bin/env python3
  2  """
  3  SSH Key Setup - Auto-detect and configure SSH keys for Sovereign OS nodes.
  4  
  5  Usage:
  6      python3 scripts/ssh_setup.py              # Check status
  7      python3 scripts/ssh_setup.py --setup      # Setup keys and authorize
  8      python3 scripts/ssh_setup.py --test       # Test connectivity
  9  """
 10  
 11  import os
 12  import sys
 13  import subprocess
 14  from pathlib import Path
 15  
 16  # Known sovereign nodes
 17  SOVEREIGN_NODES = {
 18      "nodebox": {
 19          "host": "nodebox.local",
 20          "user": "satoshi",
 21          "description": "Local dev/deployment node"
 22      }
 23  }
 24  
 25  # SSH key paths (in priority order)
 26  SSH_KEY_TYPES = [
 27      ("id_ed25519", "Ed25519 (recommended)"),
 28      ("id_rsa", "RSA"),
 29      ("id_ecdsa", "ECDSA"),
 30  ]
 31  
 32  
 33  def get_ssh_dir() -> Path:
 34      """Get SSH directory path."""
 35      return Path.home() / ".ssh"
 36  
 37  
 38  def find_existing_keys() -> list:
 39      """Find existing SSH keys."""
 40      ssh_dir = get_ssh_dir()
 41      found_keys = []
 42  
 43      for key_name, key_type in SSH_KEY_TYPES:
 44          private_key = ssh_dir / key_name
 45          public_key = ssh_dir / f"{key_name}.pub"
 46  
 47          if private_key.exists() and public_key.exists():
 48              found_keys.append({
 49                  "name": key_name,
 50                  "type": key_type,
 51                  "private": private_key,
 52                  "public": public_key,
 53                  "public_content": public_key.read_text().strip()
 54              })
 55  
 56      return found_keys
 57  
 58  
 59  def generate_key(key_type: str = "ed25519") -> dict:
 60      """Generate a new SSH key pair."""
 61      ssh_dir = get_ssh_dir()
 62      ssh_dir.mkdir(mode=0o700, exist_ok=True)
 63  
 64      key_path = ssh_dir / f"id_{key_type}"
 65  
 66      if key_path.exists():
 67          print(f"Key already exists: {key_path}")
 68          return None
 69  
 70      print(f"Generating new {key_type} key...")
 71  
 72      result = subprocess.run([
 73          "ssh-keygen",
 74          "-t", key_type,
 75          "-f", str(key_path),
 76          "-N", "",  # No passphrase
 77          "-C", f"sovereign-os@{os.uname().nodename}"
 78      ], capture_output=True, text=True)
 79  
 80      if result.returncode == 0:
 81          public_key = key_path.with_suffix(".pub")
 82          return {
 83              "name": f"id_{key_type}",
 84              "type": key_type,
 85              "private": key_path,
 86              "public": public_key,
 87              "public_content": public_key.read_text().strip()
 88          }
 89      else:
 90          print(f"Error generating key: {result.stderr}")
 91          return None
 92  
 93  
 94  def test_ssh_connection(host: str, user: str, timeout: int = 5) -> dict:
 95      """Test SSH connection to a host."""
 96      result = {
 97          "host": host,
 98          "user": user,
 99          "reachable": False,
100          "key_auth": False,
101          "error": None
102      }
103  
104      # Test with key auth (no password prompt)
105      try:
106          proc = subprocess.run([
107              "ssh",
108              "-o", "BatchMode=yes",
109              "-o", "ConnectTimeout=5",
110              "-o", "StrictHostKeyChecking=no",
111              f"{user}@{host}",
112              "echo ok"
113          ], capture_output=True, text=True, timeout=timeout + 2)
114  
115          if proc.returncode == 0:
116              result["reachable"] = True
117              result["key_auth"] = True
118          elif "Permission denied" in proc.stderr:
119              result["reachable"] = True
120              result["key_auth"] = False
121              result["error"] = "Key not authorized"
122          elif "Host key verification failed" in proc.stderr:
123              result["reachable"] = True
124              result["key_auth"] = False
125              result["error"] = "Host key changed"
126          else:
127              result["error"] = proc.stderr.strip()[:100]
128      except subprocess.TimeoutExpired:
129          result["error"] = "Connection timeout"
130      except Exception as e:
131          result["error"] = str(e)
132  
133      return result
134  
135  
136  def copy_key_to_host(public_key: str, host: str, user: str) -> bool:
137      """
138      Copy public key to remote host's authorized_keys.
139  
140      Note: This requires password auth to work initially.
141      """
142      print(f"\nTo authorize your key on {host}, run:")
143      print(f"  ssh-copy-id {user}@{host}")
144      print(f"\nOr manually add this to {user}@{host}:~/.ssh/authorized_keys:")
145      print(f"  {public_key}")
146      return False
147  
148  
149  def check_status() -> dict:
150      """Check overall SSH setup status."""
151      status = {
152          "keys": find_existing_keys(),
153          "nodes": {}
154      }
155  
156      # Test each known node
157      for node_name, node_info in SOVEREIGN_NODES.items():
158          status["nodes"][node_name] = test_ssh_connection(
159              node_info["host"],
160              node_info["user"]
161          )
162          status["nodes"][node_name]["description"] = node_info["description"]
163  
164      return status
165  
166  
167  def print_status(status: dict):
168      """Print status in human-readable format."""
169      print("═" * 60)
170      print("SSH KEY STATUS")
171      print("═" * 60)
172      print()
173  
174      # Keys
175      if status["keys"]:
176          print("Local SSH Keys:")
177          for key in status["keys"]:
178              print(f"  ✓ {key['name']} ({key['type']})")
179      else:
180          print("⚠ No SSH keys found!")
181          print("  Run: python3 scripts/ssh_setup.py --setup")
182      print()
183  
184      # Nodes
185      print("Sovereign Nodes:")
186      all_good = True
187  
188      for node_name, node_status in status["nodes"].items():
189          desc = node_status.get("description", "")
190  
191          if node_status["key_auth"]:
192              print(f"  ✓ {node_name}: Key auth working")
193          elif node_status["reachable"]:
194              all_good = False
195              print(f"  ⚠ {node_name}: Reachable but key not authorized")
196              if node_status["error"]:
197                  print(f"     Error: {node_status['error']}")
198          else:
199              print(f"  ✗ {node_name}: Not reachable")
200              if node_status["error"]:
201                  print(f"     Error: {node_status['error']}")
202  
203      print()
204  
205      if not all_good and status["keys"]:
206          print("To authorize your key on a node:")
207          print("  ssh-copy-id satoshi@nodebox.local")
208          print()
209  
210  
211  def setup():
212      """Interactive setup flow."""
213      print("═" * 60)
214      print("SSH KEY SETUP")
215      print("═" * 60)
216      print()
217  
218      # Check for existing keys
219      keys = find_existing_keys()
220  
221      if keys:
222          print("Existing SSH keys found:")
223          for key in keys:
224              print(f"  ✓ {key['name']} ({key['type']})")
225          print()
226          key = keys[0]  # Use first key
227      else:
228          print("No SSH keys found. Generating Ed25519 key...")
229          key = generate_key("ed25519")
230          if not key:
231              print("Failed to generate key.")
232              return False
233  
234      print(f"\nYour public key ({key['name']}):")
235      print(f"  {key['public_content'][:60]}...")
236      print()
237  
238      # Test nodes and offer to authorize
239      for node_name, node_info in SOVEREIGN_NODES.items():
240          print(f"\nChecking {node_name} ({node_info['host']})...")
241  
242          result = test_ssh_connection(node_info["host"], node_info["user"])
243  
244          if result["key_auth"]:
245              print(f"  ✓ Key already authorized on {node_name}")
246          elif result["reachable"]:
247              print(f"  ⚠ {node_name} reachable but key not authorized")
248              copy_key_to_host(key["public_content"], node_info["host"], node_info["user"])
249          else:
250              print(f"  ✗ {node_name} not reachable: {result.get('error', 'unknown error')}")
251  
252      print()
253      return True
254  
255  
256  def main():
257      args = sys.argv[1:]
258  
259      if "--setup" in args:
260          setup()
261      elif "--test" in args:
262          status = check_status()
263          print_status(status)
264          # Exit with error if any node has key auth issues
265          for node_status in status["nodes"].values():
266              if not node_status["key_auth"]:
267                  sys.exit(1)
268      elif "--json" in args:
269          import json
270          status = check_status()
271          # Convert Path objects to strings
272          for key in status["keys"]:
273              key["private"] = str(key["private"])
274              key["public"] = str(key["public"])
275          print(json.dumps(status, indent=2))
276      else:
277          status = check_status()
278          print_status(status)
279  
280  
281  if __name__ == "__main__":
282      main()